APIGOV-32397 AWS dataplane update

go.sum

@@ -429,8 +429,6 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
-golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
 golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
 golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -468,8 +466,6 @@ golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8=
-golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w=
 golang.org/x/mod v0.34.0 h1:xIHgNUUnW6sYkcM5Jleh05DvLOtwc6RitGHbDk4akRI=
 golang.org/x/mod v0.34.0/go.mod h1:ykgH52iCZe79kzLLMhyCUzhMci+nQj+0XkbXpNYtVjY=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -508,8 +504,6 @@ golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211209124913-491a49abca63/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
-golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
 golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
 golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -579,12 +573,8 @@ golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
-golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
 golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/telemetry v0.0.0-20260209163413-e7419c687ee4 h1:bTLqdHv7xrGlFbvf5/TXNxy/iUwwdkjhqQTJDjW7aj0=
-golang.org/x/telemetry v0.0.0-20260209163413-e7419c687ee4/go.mod h1:g5NllXBEermZrmR51cJDQxmJUHUOfRAaNyWBM+R+548=
 golang.org/x/telemetry v0.0.0-20260311193753-579e4da9a98c h1:6a8FdnNk6bTXBjR4AGKFgUKuo+7GnR3FX5L7CbveeZc=
 golang.org/x/telemetry v0.0.0-20260311193753-579e4da9a98c/go.mod h1:TpUTTEp9frx7rTdLpC9gFG9kdI7zVLFTFFlqaH2Cncw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -596,8 +586,6 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
-golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
 golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
 golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -653,8 +641,6 @@ golang.org/x/tools v0.0.0-20210105154028-b0ab187a4818/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
-golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
-golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
 golang.org/x/tools v0.43.0 h1:12BdW9CeB3Z+J/I/wj34VMl8X+fEXBxVR90JeMX5E7s=
 golang.org/x/tools v0.43.0/go.mod h1:uHkMso649BX2cZK6+RpuIPXS3ho2hZo4FVwfoy1vIk0=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

pkg/agent/events/eventlistener.go

@@ -28,7 +28,7 @@ type APIClient interface {
 // EventListener holds the various caches to save events into as they get written to the source channel.
 type EventListener struct {
 	ctx             context.Context
-	cancel          context.CancelFunc
+	cancel          context.CancelCauseFunc
 	client          APIClient
 	handlers        []handler.Handler
 	logger          log.FieldLogger
@@ -37,10 +37,10 @@ type EventListener struct {
 }
 
 // NewListenerFunc type for creating a new listener
-type NewListenerFunc func(ctx context.Context, cancel context.CancelFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, cbs ...handler.Handler) *EventListener
+type NewListenerFunc func(ctx context.Context, cancel context.CancelCauseFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, cbs ...handler.Handler) *EventListener
 
 // NewEventListener creates a new EventListener to process events based on the provided Handlers.
-func NewEventListener(ctx context.Context, cancel context.CancelFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, cbs ...handler.Handler) *EventListener {
+func NewEventListener(ctx context.Context, cancel context.CancelCauseFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, cbs ...handler.Handler) *EventListener {
 	logger := log.NewFieldLogger().
 		WithComponent("EventListener").
 		WithPackage("sdk.agent.events")
@@ -59,7 +59,7 @@ func NewEventListener(ctx context.Context, cancel context.CancelFunc, source cha
 // Stop stops the listener
 func (em *EventListener) Stop() {
 	if em != nil {
-		em.cancel()
+		em.cancel(nil)
 	}
 }
 
@@ -92,10 +92,10 @@ func (em *EventListener) start() (done bool, err error) {
 		}
 
 		if err := em.handleEvent(event); err != nil {
-			em.logger.WithError(err).Error("stream event listener error")
+			em.logger.WithError(err).Error("stream event listener error handling event")
 		}
 	case <-em.ctx.Done():
-		em.logger.Trace("stream event listener has been gracefully stopped")
+		em.logger.Trace("stream event listener context is done")
 		done = true
 		err = nil
 		break

pkg/agent/events/eventlistener_test.go

@@ -59,7 +59,7 @@ func TestEventListener_start(t *testing.T) {
 	sequenceManager := NewSequenceProvider(cacheManager, "testWatch")
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			ctx, cancel := context.WithCancel(context.Background())
+			ctx, cancel := context.WithCancelCause(context.Background())
 			listener := NewEventListener(ctx, cancel, tc.events, tc.client, sequenceManager, tc.handler)
 
 			errCh := make(chan error)
@@ -100,14 +100,14 @@ func TestEventListener_Listen(t *testing.T) {
 	cacheManager := agentcache.NewAgentCacheManager(&config.CentralConfiguration{}, false)
 	sequenceManager := NewSequenceProvider(cacheManager, "testWatch")
 	events := make(chan *proto.Event)
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancelCause(context.Background())
 	listener := NewEventListener(ctx, cancel, events, &mockAPIClient{}, sequenceManager, &mockHandler{})
 	listener.Listen()
 	listener.Stop()
 	err := ctx.Err()
 	assert.NotNil(t, err)
 
-	ctx, cancel = context.WithCancel(context.Background())
+	ctx, cancel = context.WithCancelCause(context.Background())
 	listener = NewEventListener(ctx, cancel, events, &mockAPIClient{}, sequenceManager, &mockHandler{})
 	listener.Listen()
 	close(events)
@@ -173,7 +173,7 @@ func TestEventListener_handleEvent(t *testing.T) {
 				},
 			}
 
-			ctx, cancel := context.WithCancel(context.Background())
+			ctx, cancel := context.WithCancelCause(context.Background())
 			listener := NewEventListener(ctx, cancel, make(chan *proto.Event), tc.client, sequenceManager, tc.handler)
 
 			err := listener.handleEvent(event)

pkg/agent/events/requestqueue.go

@@ -20,18 +20,18 @@ type RequestQueue interface {
 // requestQueue
 type requestQueue struct {
 	ctx       context.Context
-	cancel    context.CancelFunc
+	cancel    context.CancelCauseFunc
 	logger    log.FieldLogger
 	requestCh chan *proto.Request
 	receiveCh chan *proto.Request
 	isActive  atomic.Bool
 }
 
 // NewRequestQueueFunc type for creating a new request queue
-type NewRequestQueueFunc func(ctx context.Context, cancel context.CancelFunc, requestCh chan *proto.Request) RequestQueue
+type NewRequestQueueFunc func(ctx context.Context, cancel context.CancelCauseFunc, requestCh chan *proto.Request) RequestQueue
 
 // NewRequestQueue creates a new queue for the requests to be sent for watch subscription
-func NewRequestQueue(ctx context.Context, cancel context.CancelFunc, requestCh chan *proto.Request) RequestQueue {
+func NewRequestQueue(ctx context.Context, cancel context.CancelCauseFunc, requestCh chan *proto.Request) RequestQueue {
 	logger := log.NewFieldLogger().
 		WithComponent("requestQueue").
 		WithPackage("sdk.agent.events")
@@ -52,7 +52,7 @@ func (q *requestQueue) Stop() {
 	}
 
 	defer q.isActive.Store(false)
-	q.cancel()
+	q.cancel(nil)
 	close(q.receiveCh)
 }
 
@@ -66,8 +66,12 @@ func (q *requestQueue) Write(request *proto.Request) error {
 	}
 
 	q.logger.WithField("requestType", request.RequestType).Trace("received stream request")
-	q.receiveCh <- request
-	return nil
+	select {
+	case q.receiveCh <- request:
+		return nil
+	case <-q.ctx.Done():
+		return q.ctx.Err()
+	}
 }
 
 func (q *requestQueue) Start() {

pkg/agent/events/requestqueue_test.go

@@ -29,7 +29,7 @@ func TestRequestQueue(t *testing.T) {
 	for _, tc := range cases {
 		requestCh := make(chan *proto.Request, 1)
 		t.Run(tc.name, func(t *testing.T) {
-			ctx, cancel := context.WithCancel(context.Background())
+			ctx, cancel := context.WithCancelCause(context.Background())
 			q := NewRequestQueue(ctx, cancel, requestCh)
 			var receivedReq *proto.Request
 			wg := sync.WaitGroup{}

pkg/agent/poller/client.go

@@ -65,9 +65,9 @@ func NewPollClient(
 
 // Start the polling client
 func (p *PollClient) Start() error {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	eventCh, eventErrorCh := make(chan *proto.Event), make(chan error)
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+	eventCh := make(chan *proto.Event)
 
 	p.mutex.Lock()
 
@@ -76,7 +76,7 @@ func (p *PollClient) Start() error {
 	p.poller = p.newPollManager(p.interval, withOnStop(p.onClientStop), withHarvester(p.harvesterConfig), WithContext(ctx, cancel))
 	p.mutex.Unlock()
 	p.listener.Listen()
-	p.poller.RegisterWatch(eventCh, eventErrorCh)
+	p.poller.RegisterWatch(eventCh)
 
 	if p.onStreamConnection != nil {
 		p.onStreamConnection()
@@ -86,12 +86,11 @@ func (p *PollClient) Start() error {
 	p.initialized = true
 	p.mutex.Unlock()
 
-	select {
-	case err := <-eventErrorCh:
-		return err
-	case <-ctx.Done():
-		return fmt.Errorf("poll client context has been closed")
+	<-ctx.Done()
+	if cause := context.Cause(ctx); cause != nil {
+		return cause
 	}
+	return fmt.Errorf("poll client context has been closed")
 }
 
 // Status returns an error if the poller is not running

pkg/agent/poller/options.go

@@ -53,7 +53,7 @@ func withOnStop(cb onClientStopCb) executorOpt {
 	}
 }
 
-func WithContext(ctx context.Context, cancel context.CancelFunc) executorOpt {
+func WithContext(ctx context.Context, cancel context.CancelCauseFunc) executorOpt {
 	return func(m *pollExecutor) {
 		m.ctx = ctx
 		m.cancel = cancel

pkg/agent/poller/poller.go

@@ -14,7 +14,7 @@ import (
 
 type pollExecutor struct {
 	ctx           context.Context
-	cancel        context.CancelFunc
+	cancel        context.CancelCauseFunc
 	harvester     harvester.Harvest
 	sequence      events.SequenceProvider
 	topicSelfLink string
@@ -33,7 +33,7 @@ func newPollExecutor(interval time.Duration, options ...executorOpt) *pollExecut
 		WithComponent("pollExecutor").
 		WithPackage("sdk.agent.poller")
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancelCause(context.Background())
 	pm := &pollExecutor{
 		logger:   logger,
 		timer:    time.NewTimer(interval),
@@ -50,32 +50,28 @@ func newPollExecutor(interval time.Duration, options ...executorOpt) *pollExecut
 }
 
 // RegisterWatch registers a watch topic for polling events and publishing events on a channel
-func (m *pollExecutor) RegisterWatch(eventChan chan *proto.Event, errChan chan error) {
+func (m *pollExecutor) RegisterWatch(eventChan chan *proto.Event) {
 	m.logger.Trace("register watch topic for polling and publishing events")
 	if m.harvester == nil {
-		go func() {
-			m.Stop()
-			errChan <- fmt.Errorf("harvester is not configured for the polling client")
-		}()
+		err := fmt.Errorf("harvester is not configured for the polling client")
+		m.cancel(err)
+		m.Stop()
 		return
 	}
 
 	if m.sequence.GetSequence() < 0 {
 		m.onHarvesterErr()
-		go func() {
-			m.Stop()
-			errChan <- fmt.Errorf("do not have a sequence id, stopping poller")
-		}()
+		err := fmt.Errorf("do not have a sequence id, stopping poller")
+		m.cancel(err)
+		m.Stop()
 		return
 	}
 
 	if err := m.harvester.EventCatchUp(m.ctx, m.topicSelfLink, eventChan); err != nil {
 		m.logger.WithError(err).Error("harvester returned an error when syncing events")
 		m.onHarvesterErr()
-		go func() {
-			m.Stop()
-			errChan <- err
-		}()
+		m.cancel(err)
+		m.Stop()
 		return
 	}
 
@@ -84,9 +80,10 @@ func (m *pollExecutor) RegisterWatch(eventChan chan *proto.Event, errChan chan e
 	m.lock.Unlock()
 
 	go func() {
-		err := m.sync(m.topicSelfLink, eventChan)
+		if err := m.sync(m.topicSelfLink, eventChan); err != nil {
+			m.cancel(err)
+		}
 		m.Stop()
-		errChan <- err
 	}()
 }
 
@@ -142,7 +139,7 @@ func (m *pollExecutor) onHarvesterErr() {
 // Stop stops the poller
 func (m *pollExecutor) Stop() {
 	m.timer.Stop()
-	m.cancel()
+	m.cancel(nil)
 
 	m.lock.Lock()
 	defer m.lock.Unlock()