add redis queue & async message strategy #615
Merged
Conversation
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
yanjiaxin534
requested review from
Haishi2016 and
msftcoderdjw
as code owners
upsert/delete/get with metadata
yanjiaxin534
force-pushed
the
remote-agent-redis
branch
from
fb0d691 to
1e4466b
Compare
yanjiaxin534
force-pushed
the
remote-agent-redis
branch
from
a75a2cd to
c0c5746
Compare
msftcoderdjw
approved these changes
Jan 27, 2025
| @@ -65,6 +65,11 @@ func toMemoryQueueProviderConfig(config providers.IProviderConfig) (MemoryQueueP | |||
| return ret, err | |||
| } | |||
|
|
|||
| // fake | |||
| func (s *MemoryQueueProvider) QueryByPaging(queueName string, start string, size int) ([][]byte, string, error) { | |||
| return [][]byte{}, "", nil | |||
There was a problem hiding this comment.
Better throw error for not implemented.
| assert.Equal(t, v1alpha2.OK, resp.State) | ||
| json.Unmarshal(resp.Body, &summary) | ||
| assert.False(t, summary.Skipped) | ||
| // // deploy |
There was a problem hiding this comment.
Why we comment out these cases?
| @@ -76,14 +81,18 @@ func (s *MemoryQueueProvider) Init(config providers.IProviderConfig) error { | |||
| return nil | |||
| } | |||
|
|
|||
| func (s *MemoryQueueProvider) Enqueue(queue string, data interface{}) error { | |||
| // fake | |||
| func (s *MemoryQueueProvider) RemoveFromQueue(queue string, messageID string) error { | |||
| @@ -7,8 +7,10 @@ | |||
| package queue | |||
|
|
|||
| type IQueueProvider interface { | |||
| Enqueue(queue string, element interface{}) error | |||
| Enqueue(queue string, element interface{}) (string, error) | |||
There was a problem hiding this comment.
What does string return value mean?
For IQueueProvider, pass context in.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Roles of Each Topic:
DeploymentPlanTopic:
Splits the plan into sub-steps and sends them to DeploymentStepTopic.
DeploymentStepTopic:
Responsible for executing steps.
If the step's target is not a remote target, it executes the step and sends the result to CollectStepResultTopic.
If the step's target is a remote target, it sends the step execution information to the Redis queue to be executed by the remote agent.
CollectStepResultTopic:
Receives step execution results, which may come from either the remote agent or local execution.
Execution Details:
Each time when DeploymentPlanTopic receives a plan:
First -> splits the plan. Then sends the first step (step 0) to DeploymentStepTopic.
When DeploymentStepTopic receives a message:
If the target is local, it executes the step and sends the result to CollectStepResultTopic.
If the target is remote, it sends the step to the queue and waits for it to be picked up.
When CollectStepResultTopic receives a result, it checks if it is the last step:
If it is the last step, it executes the function to handle the end of the plan.
If it is not the last step, it sends the next step to DeploymentStepTopic.
Handling the End of the Plan:
Remote Agent Message Interfaces:
solution/tasks -> onGetRequest:
Retrieves tasks by checking the target name, namespace, and whether to get all.
solution/task/getResult -> onGetResponse:
Receives the result from the remote agent, parses it, and sends it to CollectStepResultTopic.
Data Structure
Operation Body:
Responsible for storing parameters related to the operationId. When a result is received from the remote agent:
It first queries the related parameters based on the operationId, then returns the result, and finally deletes the information corresponding to that operationId.
The benefit of this approach is that if the remote agent sends two identical requests, it will not send the result twice to CollectStepResultTopic. Additionally, random requests from the remote agent will not be processed.