generated from steadybit/extension-scaffold
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
277 additions
and
25 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,169 @@ | ||
| /* | ||
| * Copyright 2023 steadybit GmbH. All rights reserved. | ||
| */ | ||
|
|
||
| package extkafka | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "github.com/google/uuid" | ||
| "github.com/rs/zerolog/log" | ||
| "github.com/steadybit/action-kit/go/action_kit_api/v2" | ||
| "github.com/steadybit/action-kit/go/action_kit_sdk" | ||
| "github.com/steadybit/extension-kit/extbuild" | ||
| "github.com/steadybit/extension-kit/extutil" | ||
| ) | ||
|
|
||
| type consumeMessageActionFixedAmount struct{} | ||
|
|
||
| // Make sure Action implements all required interfaces | ||
| var ( | ||
| _ action_kit_sdk.Action[KafkaBrokerAttackState] = (*consumeMessageActionFixedAmount)(nil) | ||
| _ action_kit_sdk.ActionWithStatus[KafkaBrokerAttackState] = (*consumeMessageActionFixedAmount)(nil) | ||
|
|
||
| _ action_kit_sdk.ActionWithStop[KafkaBrokerAttackState] = (*consumeMessageActionFixedAmount)(nil) | ||
| ) | ||
|
|
||
| func NewConsumeMessageActionFixedAmount() action_kit_sdk.Action[KafkaBrokerAttackState] { | ||
| return &consumeMessageActionFixedAmount{} | ||
| } | ||
|
|
||
| func (l *consumeMessageActionFixedAmount) NewEmptyState() KafkaBrokerAttackState { | ||
| return KafkaBrokerAttackState{} | ||
| } | ||
|
|
||
| // Describe returns the action description for the platform with all required information. | ||
| func (l *consumeMessageActionFixedAmount) Describe() action_kit_api.ActionDescription { | ||
| return action_kit_api.ActionDescription{ | ||
| Id: TargetIDConsumeFixedAmount, | ||
| Label: "Consume X records", | ||
| Description: "Consume a certain amount of kafka messages for a given duration", | ||
| Version: extbuild.GetSemverVersionStringOrUnknown(), | ||
| Icon: extutil.Ptr(kafkaMessageFixedAmount), | ||
| TargetSelection: extutil.Ptr(action_kit_api.TargetSelection{ | ||
| TargetType: kafkaTopicTargetId, | ||
| SelectionTemplates: extutil.Ptr([]action_kit_api.TargetSelectionTemplate{ | ||
| { | ||
| Label: "by topic name", | ||
| Description: extutil.Ptr("Find topic by name"), | ||
| Query: "kafka.topic.name=\"\"", | ||
| }, | ||
| }), | ||
| }), | ||
| //Widgets: extutil.Ptr([]action_kit_api.Widget{ | ||
| // action_kit_api.PredefinedWidget{ | ||
| // Type: action_kit_api.ComSteadybitWidgetPredefined, | ||
| // PredefinedWidgetId: "com.steadybit.widget.predefined.HttpCheck", | ||
| // }, | ||
| //}), | ||
|
|
||
| // Technology for the targets to appear in | ||
| Technology: extutil.Ptr("Kafka"), | ||
|
|
||
| // To clarify the purpose of the action: | ||
| // Check: Will perform checks on the targets | ||
| Kind: action_kit_api.Check, | ||
|
|
||
| // How the action is controlled over time. | ||
| // External: The agent takes care and calls stop then the time has passed. Requires a duration parameter. Use this when the duration is known in advance. | ||
| // Internal: The action has to implement the status endpoint to signal when the action is done. Use this when the duration is not known in advance. | ||
| // Instantaneous: The action is done immediately. Use this for actions that happen immediately, e.g. a reboot. | ||
| TimeControl: action_kit_api.TimeControlInternal, | ||
|
|
||
| // The parameters for the action | ||
| Parameters: []action_kit_api.ActionParameter{ | ||
| //------------------------ | ||
| // Request Definition | ||
| //------------------------ | ||
| { | ||
| Name: "-", | ||
| Label: "-", | ||
| Type: action_kit_api.Separator, | ||
| Order: extutil.Ptr(5), | ||
| }, | ||
| //------------------------ | ||
| // Repitions | ||
| //------------------------ | ||
| repetitionControl, | ||
| { | ||
| Name: "numberOfRecords", | ||
| Label: "Number of Records.", | ||
| Description: extutil.Ptr("Fixed number of Records, distributed to given duration"), | ||
| Type: action_kit_api.Integer, | ||
| Required: extutil.Ptr(true), | ||
| DefaultValue: extutil.Ptr("1"), | ||
| Order: extutil.Ptr(7), | ||
| }, | ||
| duration, | ||
| { | ||
| Name: "-", | ||
| Label: "-", | ||
| Type: action_kit_api.Separator, | ||
| Order: extutil.Ptr(9), | ||
| }, | ||
| //------------------------ | ||
| // Result Verification | ||
| //------------------------ | ||
| resultVerification, | ||
| successRate, | ||
|
|
||
| //------------------------ | ||
| // Additional Settings | ||
| //------------------------ | ||
|
|
||
| maxConcurrent, | ||
| }, | ||
| Status: extutil.Ptr(action_kit_api.MutatingEndpointReferenceWithCallInterval{ | ||
| CallInterval: extutil.Ptr("1s"), | ||
| }), | ||
| Stop: extutil.Ptr(action_kit_api.MutatingEndpointReference{}), | ||
| } | ||
| } | ||
|
|
||
| func (l *consumeMessageActionFixedAmount) Prepare(_ context.Context, state *KafkaBrokerAttackState, request action_kit_api.PrepareActionRequestBody) (*action_kit_api.PrepareResult, error) { | ||
| if extutil.ToInt64(request.Config["duration"]) == 0 { | ||
| return nil, errors.New("duration must be greater than 0") | ||
| } | ||
| state.DelayBetweenRequestsInMS = getDelayBetweenRequestsInMsFixedAmount(extutil.ToInt64(request.Config["duration"]), extutil.ToInt64(request.Config["numberOfRecords"])) | ||
| state.Topic = extutil.MustHaveValue(request.Target.Attributes, "kafka.topic.name")[0] | ||
| return prepare(false, request, state, checkEndedFixedAmount) | ||
| } | ||
|
|
||
| // Start is called to start the action | ||
| // You can mutate the state here. | ||
| // You can use the result to return messages/errors/metrics or artifacts | ||
| func (l *consumeMessageActionFixedAmount) Start(_ context.Context, state *KafkaBrokerAttackState) (*action_kit_api.StartResult, error) { | ||
| start(state) | ||
| return nil, nil | ||
| } | ||
|
|
||
| // Status is called to get the current status of the action | ||
| func (l *consumeMessageActionFixedAmount) Status(_ context.Context, state *KafkaBrokerAttackState) (*action_kit_api.StatusResult, error) { | ||
| executionRunData, err := loadExecutionRunData(state.ExecutionID) | ||
| if err != nil { | ||
| log.Error().Err(err).Msg("Failed to load execution run data") | ||
| return nil, err | ||
| } | ||
|
|
||
| completed := checkEndedFixedAmount(executionRunData, state) | ||
| if completed { | ||
| stopTickers(executionRunData) | ||
| log.Info().Msg("Action completed") | ||
| } | ||
|
|
||
| latestMetrics := retrieveLatestMetrics(executionRunData.metrics) | ||
|
|
||
| return &action_kit_api.StatusResult{ | ||
| Completed: completed, | ||
| Metrics: extutil.Ptr(latestMetrics), | ||
| }, nil | ||
| } | ||
|
|
||
| func (l *consumeMessageActionFixedAmount) Stop(_ context.Context, state *KafkaBrokerAttackState) (*action_kit_api.StopResult, error) { | ||
| return stop(state) | ||
| } | ||
|
|
||
| func (l *consumeMessageActionFixedAmount) getExecutionRunData(executionID uuid.UUID) (*ExecutionRunData, error) { | ||
| return loadExecutionRunData(executionID) | ||
| } |
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.