-
Notifications
You must be signed in to change notification settings - Fork 13
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
APIGOV-29256 - agent status update over gRPC stream (#861)
* APIGOV-29256 - new format for User-Agent header * APIGOV-29256 - watch proto updates - added optional request type to watch request to identify token refresh or agent status update. Defaults to token refresh for backward compatibility - added TriggerAction event type to watch event - added optional action metadata to watch event metadata to hold details about TriggerAction event * APIGOV-29256 - update user agent in sample * APIGOV-29256 - update user agent in sample * APIGOV-29256 - updated watch client for forwarding request using a channel * APIGOV-29256 - updated stream client to use request queue to send status updates * APIGOV-29256 - use stream to send status update when using gRPC * APIGOV-29256 - unit test fixes * APIGOV-29256 - code review update * APIGOV-29256 - fix regex to parse commit sha * APIGOV-29256 - agent dataplane type to component map includes fix to close wait on error before any watch request send * APIGOV-29256 - race condition fix * APIGOV-29256 - Fix user-agent regex - thanks Jason * APIGOV-29256 - watch client test optimization
- Loading branch information
1 parent
99e67fb
commit 2d68f3b
Showing
21 changed files
with
1,217 additions
and
210 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| package events | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "sync" | ||
|
|
||
| "github.com/Axway/agent-sdk/pkg/util/log" | ||
|
|
||
| "github.com/Axway/agent-sdk/pkg/watchmanager/proto" | ||
| ) | ||
|
|
||
| type RequestQueue interface { | ||
| Start() | ||
| Write(request *proto.Request) error | ||
| Stop() | ||
| IsActive() bool | ||
| } | ||
|
|
||
| // requestQueue | ||
| type requestQueue struct { | ||
| cancel context.CancelFunc | ||
| ctx context.Context | ||
| logger log.FieldLogger | ||
| requestCh chan *proto.Request | ||
| receiveCh chan *proto.Request | ||
| isActive bool | ||
| lock *sync.Mutex | ||
| } | ||
|
|
||
| // NewRequestQueueFunc type for creating a new request queue | ||
| type NewRequestQueueFunc func(requestCh chan *proto.Request) RequestQueue | ||
|
|
||
| // NewRequestQueue creates a new queue for the requests to be sent for watch subscription | ||
| func NewRequestQueue(requestCh chan *proto.Request) RequestQueue { | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| logger := log.NewFieldLogger(). | ||
| WithComponent("requestQueue"). | ||
| WithPackage("sdk.agent.events") | ||
|
|
||
| return &requestQueue{ | ||
| cancel: cancel, | ||
| ctx: ctx, | ||
| logger: logger, | ||
| requestCh: requestCh, | ||
| receiveCh: make(chan *proto.Request, 1), | ||
| lock: &sync.Mutex{}, | ||
| } | ||
| } | ||
|
|
||
| func (q *requestQueue) Stop() { | ||
| q.lock.Lock() | ||
| defer q.lock.Unlock() | ||
|
|
||
| if q.cancel != nil { | ||
| q.cancel() | ||
| } | ||
| } | ||
|
|
||
| func (q *requestQueue) IsActive() bool { | ||
| q.lock.Lock() | ||
| defer q.lock.Unlock() | ||
|
|
||
| return q.isActive | ||
| } | ||
|
|
||
| func (q *requestQueue) Write(request *proto.Request) error { | ||
| q.lock.Lock() | ||
| defer q.lock.Unlock() | ||
|
|
||
| if !q.isActive { | ||
| return errors.New("request queue is not active") | ||
| } | ||
|
|
||
| if q.receiveCh != nil { | ||
| q.logger.WithField("requestType", request.RequestType).Trace("received stream request") | ||
| q.receiveCh <- request | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (q *requestQueue) Start() { | ||
| go func() { | ||
| q.lock.Lock() | ||
| q.isActive = true | ||
| q.lock.Unlock() | ||
|
|
||
| defer func() { | ||
| q.lock.Lock() | ||
| defer q.lock.Unlock() | ||
| q.isActive = false | ||
| }() | ||
|
|
||
| for { | ||
| if q.process() { | ||
| break | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| func (q *requestQueue) process() bool { | ||
| done := false | ||
| select { | ||
| case req := <-q.receiveCh: | ||
| q.logger.WithField("requestType", req.RequestType).Trace("forwarding stream request") | ||
| q.requestCh <- req | ||
| q.logger.WithField("requestType", req.RequestType).Trace("stream request forwarded") | ||
| case <-q.ctx.Done(): | ||
| q.logger.Trace("stream request queue has been gracefully stopped") | ||
| done = true | ||
| if q.receiveCh != nil { | ||
| close(q.receiveCh) | ||
| q.receiveCh = nil | ||
| } | ||
| break | ||
| } | ||
|
|
||
| return done | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| package events | ||
|
|
||
| import ( | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/Axway/agent-sdk/pkg/watchmanager/proto" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestRequestQueue(t *testing.T) { | ||
| cases := []struct { | ||
| name string | ||
| writeErr bool | ||
| queueActivated bool | ||
| }{ | ||
| { | ||
| name: "write error when queue is not active", | ||
| writeErr: true, | ||
| queueActivated: false, | ||
| }, | ||
| { | ||
| name: "write success with active queue", | ||
| queueActivated: true, | ||
| }, | ||
| } | ||
| for _, tc := range cases { | ||
| requestCh := make(chan *proto.Request, 1) | ||
| t.Run(tc.name, func(t *testing.T) { | ||
| q := NewRequestQueue(requestCh) | ||
| var receivedReq *proto.Request | ||
| wg := sync.WaitGroup{} | ||
| if tc.queueActivated { | ||
| wg.Add(1) | ||
| q.Start() | ||
| time.Sleep(1 * time.Second) | ||
| go func() { | ||
| receivedReq = <-requestCh | ||
| q.Stop() | ||
| wg.Done() | ||
| }() | ||
| } | ||
|
|
||
| req := &proto.Request{ | ||
| RequestType: proto.RequestType_AGENT_STATUS.Enum(), | ||
| AgentStatus: &proto.AgentStatus{ | ||
| State: "running", | ||
| }, | ||
| } | ||
| writeErr := q.Write(req) | ||
| if tc.writeErr { | ||
| assert.NotNil(t, writeErr) | ||
| return | ||
| } | ||
| assert.Nil(t, writeErr) | ||
|
|
||
| wg.Wait() | ||
| assert.Equal(t, req, receivedReq) | ||
| assert.False(t, q.IsActive()) | ||
| }) | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.