Unhandled sftp error

[facorazza]

I'm using the sftp input but I get the following error:

/usr/local/go/src/sync/cond.go:70 +0x85 fp=0xc000ece758 sp=0xc000ece718 pc=0x47bc45
golang.org/x/crypto/ssh.(*mux).Wait(0xc00134fb90)
/go/pkg/mod/golang.org/x/[email protected]/ssh/mux.go:110 +0x8e fp=0xc000ece7a8 sp=0xc000ece758 pc=0x360fd8e
golang.org/x/crypto/ssh.(*connection).Wait(0x0?)
<autogenerated>:1 +0x25 fp=0xc000ece7c0 sp=0xc000ece7a8 pc=0x3617845
golang.org/x/crypto/ssh.NewClient.func1()
/go/pkg/mod/golang.org/x/[email protected]/ssh/client.go:62 +0x28 fp=0xc000ece7e0 sp=0xc000ece7c0 pc=0x35f65e8
runtime.goexit()
/usr/local/go/src/runtime/asm_amd64.s:1650 +0x1 fp=0xc000ece7e8 sp=0xc000ece7e0 pc=0x472221
created by golang.org/x/crypto/ssh.NewClient in goroutine 41
/go/pkg/mod/golang.org/x/[email protected]/ssh/client.go:61 +0x165
goroutine 146 [chan receive, 96 minutes]:
runtime.gopark(0xc0023b6e10?, 0xc0023b6f78?, 0x90?, 0x6f?, 0xc0023b6fc0?)
/usr/local/go/src/runtime/proc.go:398 +0xce fp=0xc0023a0e68 sp=0xc0023a0e48 pc=0x43ef0e
runtime.chanrecv(0xc001f3d380, 0xc0023a0f50, 0x1)
/usr/local/go/src/runtime/chan.go:583 +0x3cd fp=0xc0023a0ee0 sp=0xc0023a0e68 pc=0x409bed
runtime.chanrecv2(0xc0023cc630?, 0xc0023b73b0?)
/usr/local/go/src/runtime/chan.go:447 +0x12 fp=0xc0023a0f08 sp=0xc0023a0ee0 pc=0x409812
golang.org/x/crypto/ssh.(*Session).wait(0xc0023cca20?, 0xc0023b7608?)
/go/pkg/mod/golang.org/x/[email protected]/ssh/session.go:423 +0x65 fp=0xc0023a0fa8 sp=0xc0023a0f08 pc=0x3611785
golang.org/x/crypto/ssh.newSession.func1()
/go/pkg/mod/golang.org/x/[email protected]/ssh/session.go:593 +0x25 fp=0xc0023a0fe0 sp=0xc0023a0fa8 pc=0x3611c05
runtime.goexit()
/usr/local/go/src/runtime/asm_amd64.s:1650 +0x1 fp=0xc0023a0fe8 sp=0xc0023a0fe0 pc=0x472221
created by golang.org/x/crypto/ssh.newSession in goroutine 41
/go/pkg/mod/golang.org/x/[email protected]/ssh/session.go:592 +0xf1
goroutine 147 [sync.Cond.Wait]:
runtime.gopark(0xc00253b200?, 0x410805?, 0x98?, 0x9c?, 0x418a25?)
/usr/local/go/src/runtime/proc.go:398 +0xce fp=0xc002259c30 sp=0xc002259c10 pc=0x43ef0e
runtime.goparkunlock(...)
/usr/local/go/src/runtime/proc.go:404
sync.runtime_notifyListWait(0xc0021b32d0, 0xf3ce6a)
/usr/local/go/src/runtime/sema.go:527 +0x159 fp=0xc002259c80 sp=0xc002259c30 pc=0x46e439
sync.(*Cond).Wait(0x7f9b20a7cc00?)
/usr/local/go/src/sync/cond.go:70 +0x85 fp=0xc002259cc0 sp=0xc002259c80 pc=0x47bc45
golang.org/x/crypto/ssh.(*buffer).Read(0xc00210e1a0, {0xc005845970, 0x4, 0x4})
/go/pkg/mod/golang.org/x/[email protected]/ssh/buffer.go:94 +0x1fb fp=0xc002259d48 sp=0xc002259cc0 pc=0x35edffb
golang.org/x/crypto/ssh.(*channel).ReadExtended(0xc00105d8c0, {0xc005845970?, 0xc002259e20?, 0x410805?}, 0x40909e?)
/go/pkg/mod/golang.org/x/[email protected]/ssh/channel.go:363 +0x93 fp=0xc002259da8 sp=0xc002259d48 pc=0x35f17f3
golang.org/x/crypto/ssh.(*channel).Read(0x4091d3?, {0xc005845970?, 0x7f9b1fee06c8?, 0x7f9b6814af18?})
/go/pkg/mod/golang.org/x/[email protected]/ssh/channel.go:540 +0x25 fp=0xc002259de0 sp=0xc002259da8 pc=0x35f2625
github.com/pkg/sftp.(*conn).Read(0xc005845970?, {0xc005845970?, 0xc002259e48?, 0x454069?})
<autogenerated>:1 +0x29 fp=0xc002259e10 sp=0xc002259de0 pc=0x362e3c9
io.ReadAtLeast({0x6603b80, 0xc002302090}, {0xc005845970, 0x4, 0x4}, 0x4)
/usr/local/go/src/io/io.go:335 +0x90 fp=0xc002259e58 sp=0xc002259e10 pc=0x4e7490
io.ReadFull(...)
/usr/local/go/src/io/io.go:354
github.com/pkg/sftp.recvPacket({0x6603b80, 0xc002302090}, 0x0, 0x0?)
/go/pkg/mod/github.com/pkg/[email protected]/packet.go:277 +0x99 fp=0xc002259ed0 sp=0xc002259e58 pc=0x3628379
github.com/pkg/sftp.(*conn).recvPacket(...)
/go/pkg/mod/github.com/pkg/[email protected]/conn.go:25
github.com/pkg/sftp.(*clientConn).recv(0xc002302090)
/go/pkg/mod/github.com/pkg/[email protected]/conn.go:72 +0xd4 fp=0xc002259f90 sp=0xc002259ed0 pc=0x3625b54
github.com/pkg/sftp.NewClientPipe.func1()
/go/pkg/mod/github.com/pkg/[email protected]/client.go:242 +0x58 fp=0xc002259fe0 sp=0xc002259f90 pc=0x361bc18
runtime.goexit()
/usr/local/go/src/runtime/asm_amd64.s:1650 +0x1 fp=0xc002259fe8 sp=0xc002259fe0 pc=0x472221
created by github.com/pkg/sftp.NewClientPipe in goroutine 41
/go/pkg/mod/github.com/pkg/[email protected]/client.go:239 +0x368

[gregfurman]

Hey @facorazza! Thanks for raising this here. Couple of questions:

1. Which version of Bento are you using?
2. What OS/Architecture are you running on? This looks like it could be the source of the issue.
3. Can you share any config? I'm curious if this is happening at connect time or when data is being ingested.

[gregfurman]

FYI, it appears the above issue is related to this regression in the sftp library pkg/sftp#603

Can you explain the scenario where you're seeing this error occur? It'll help me figure out where to properly address this deadlock issue.

[facorazza]

Hey @facorazza! Thanks for raising this here. Couple of questions:

1. Which version of Bento are you using?

2. What OS/Architecture are you running on? This looks like it could be the source of the issue.

3. Can you share any config? I'm curious if this is happening at connect time or when data is being ingested.

Hey @gregfurman, sorry for the late reply.

1. I was on version 1.13.0 and have since updated bento to version 1.14.1 and I'm getting the same error.
2. I'm running the container version against an sftp server hosted on Rocky linux 9. Both run on metal that has the x86 architecture.
3. I'm unable to replicate the error consistently. It seems to happen every so often, but Kubernetes takes care of it and creates a new instance. The problem I'm facing with the sftp is that bento seems unable to initialize the input if there are no files that match the blob path. Here's the config:

  input:
    label: sftp
    sftp:
      address: '${SFTP_ADDRESS}'
      credentials:
        username: '${SFTP_USERNAME}'
        password: '${SFTP_PASSWORD}'
      paths: ['data/out/*.csv']
      auto_replay_nacks: true
      delete_on_finish: true
      scanner:
        to_the_end: {}
      watcher:
        enabled: true
        minimum_age: 30s
        poll_interval: 5s
        cache: "transfers"

  cache_resources:
    - label: "transfers"
      memory:
        default_ttl: 5m
        compaction_interval: 60s
        init_values: {}

Logs when no matching files are available.

level=info msg="Running main config from specified file" @service=bento bento_version=v1.4.1 path=/bento.yaml
level=info msg="Listening for HTTP requests at: http://0.0.0.0:4195" @service=bento
level=info msg="Launching a Bento instance, use CTRL+C to close" @service=bento
level=info msg="Output type aws_s3 is now active" @service=bento label=mps_s3 path=root.output

[gregfurman]

Thanks for the info!

The problem I'm facing with the sftp is that bento seems unable to initialize the input if there are no files that match the blob path.

As far as I'm aware, the watcher should circumvent this file not existing issue so will dig into this a bit more. Any chance you can set your log level to DEBUG and provide some more logs for when you start up?

[facorazza]

Unfortunately, the log level is already set to DEBUG

[gregfurman]

I think the input is blocking until a file gets added, after which it should probably initialise. What happens if you start up Bento and then add a file to that path? Does it trigger anything?

Will also try recreate this on my end and get back to you.