Skip to content

NTHv2 core functionality #612

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Apr 14, 2022
Merged

NTHv2 core functionality #612

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
ac0ce04
core functionality
cjerad Apr 1, 2022
68b973e
fix capitalization
cjerad Apr 4, 2022
2beddc7
add more ways to specify AWS region
cjerad Apr 4, 2022
c4ee5fb
tidy up interface organization
cjerad Apr 4, 2022
0407de6
remove SQS request params from Terminator spec
cjerad Apr 5, 2022
69a0544
update copyright attribution
cjerad Apr 6, 2022
17c2ad0
remove boilerplate comment
cjerad Apr 6, 2022
d217816
expand "Addr" to "Address"
cjerad Apr 6, 2022
87dccd7
fix capitalization
cjerad Apr 6, 2022
6a7b44b
rename event files
cjerad Apr 7, 2022
ab10afa
add more logging for parsing failures
cjerad Apr 7, 2022
186caec
dedup lifecycle action completion code
cjerad Apr 7, 2022
ef556fc
add more test cases
cjerad Apr 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
fix capitalization 
* "Url" to "URL"
* "Aws" to "AWS"
* "Asg" to "ASG"
* "Ec2" to "EC2"
* "Sqs" to "SQS"
  • Loading branch information
@cjerad
cjerad committed Apr 4, 2022
commit 68b973e27233612dc34bdf9a5fedec05cf9aa776
6 changes: 3 additions & 3 deletions src/api/v1alpha1/terminator_logging.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,12 @@ import (
)

func (t *TerminatorSpec) MarshalLogObject(enc zapcore.ObjectEncoder) error {
enc.AddObject("sqs", t.Sqs)
enc.AddObject("sqs", t.SQS)
enc.AddObject("drain", t.Drain)
return nil
}

func (s SqsSpec) MarshalLogObject(enc zapcore.ObjectEncoder) error {
func (s SQSSpec) MarshalLogObject(enc zapcore.ObjectEncoder) error {
enc.AddArray("attributeNames", zapcore.ArrayMarshalerFunc(func(enc zapcore.ArrayEncoder) error {
for _, attrName := range s.AttributeNames {
enc.AppendString(attrName)
Expand All @@ -43,7 +43,7 @@ func (s SqsSpec) MarshalLogObject(enc zapcore.ObjectEncoder) error {
return nil
}))

enc.AddString("queueUrl", s.QueueUrl)
enc.AddString("queueURL", s.QueueURL)
enc.AddInt64("visibilityTimeoutSeconds", s.VisibilityTimeoutSeconds)
enc.AddInt64("waitTimeSeconds", s.WaitTimeSeconds)
return nil
Expand Down
8 changes: 4 additions & 4 deletions src/api/v1alpha1/terminator_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,17 +30,17 @@ type TerminatorSpec struct {
// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
// Important: Run "make" to regenerate code after modifying this file

Sqs SqsSpec `json:"sqs,omitempty"`
SQS SQSSpec `json:"sqs,omitempty"`
Drain DrainSpec `json:"drain,omitempty"`
}

// SqsSpec defines inputs to SQS "receive messages" requests.
type SqsSpec struct {
// SQSSpec defines inputs to SQS "receive messages" requests.
type SQSSpec struct {
// https://pkg.go.dev/github.com/aws/[email protected]/service/sqs#ReceiveMessageInput
AttributeNames []string `json:"attributeNames,omitempty"`
MaxNumberOfMessages int64 `json:"maxNumberOfMessages,omitempty"`
MessageAttributeNames []string `json:"messageAttributeNames,omitempty"`
QueueUrl string `json:"queueUrl,omitempty"`
QueueURL string `json:"queueURL,omitempty"`
VisibilityTimeoutSeconds int64 `json:"visibilityTimeoutSeconds,omitempty"`
WaitTimeSeconds int64 `json:"waitTimeSeconds,omitempty"`
}
Expand Down
12 changes: 6 additions & 6 deletions src/api/v1alpha1/terminator_validation.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ import (

var (
// https://github.com/aws/aws-sdk-go/blob/v1.38.55/service/sqs/api.go#L3966-L3994
knownSqsAttributeNames = sets.NewString(sqs.MessageSystemAttributeName_Values()...)
knownSQSAttributeNames = sets.NewString(sqs.MessageSystemAttributeName_Values()...)
)

func (t *Terminator) Validate(_ context.Context) (errs *apis.FieldError) {
Expand All @@ -41,12 +41,12 @@ func (t *Terminator) Validate(_ context.Context) (errs *apis.FieldError) {
}

func (t *TerminatorSpec) validate() (errs *apis.FieldError) {
return t.Sqs.validate().ViaField("sqs")
return t.SQS.validate().ViaField("sqs")
}

func (s *SqsSpec) validate() (errs *apis.FieldError) {
func (s *SQSSpec) validate() (errs *apis.FieldError) {
for _, attrName := range s.AttributeNames {
if !knownSqsAttributeNames.Has(attrName) {
if !knownSQSAttributeNames.Has(attrName) {
errs = errs.Also(apis.ErrInvalidValue(attrName, "attributeNames"))
}
}
Expand Down Expand Up @@ -74,8 +74,8 @@ func (s *SqsSpec) validate() (errs *apis.FieldError) {
}
}

if _, err := url.Parse(s.QueueUrl); err != nil {
errs = errs.Also(apis.ErrInvalidValue(s.QueueUrl, "queueUrl", "must be a valid URL"))
if _, err := url.Parse(s.QueueURL); err != nil {
errs = errs.Also(apis.ErrInvalidValue(s.QueueURL, "queueURL", "must be a valid URL"))
}

if s.VisibilityTimeoutSeconds < 0 {
Expand Down
10 changes: 5 additions & 5 deletions src/api/v1alpha1/zz_generated.deepcopy.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 2 additions & 2 deletions src/charts/aws-node-termination-handler-2/templates/node.k8s.aws_terminators.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,13 +116,13 @@ spec:
default:
{{- toYaml . | nindent 22 }}
{{- end }}
queueUrl:
queueURL:
description: |
The URL of the Amazon SQS queue from which messages are received.

* Queue URLs and names are case-sensitive.

* QueueUrl is a required field
* QueueURL is a required field
type: string
visibilityTimeoutSeconds:
description: |
Expand Down
12 changes: 6 additions & 6 deletions src/cmd/controller/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ func main() {
}
kubeClient := mgr.GetClient()

awsSession, err := newAwsSession()
awsSession, err := newAWSSession()
if err != nil {
logger.With("error", err).Fatal("failed to initialize AWS session")
}
Expand Down Expand Up @@ -161,7 +161,7 @@ func main() {
if err != nil {
logger.With("error", err).Fatal("failed to create ASG instance-terminate lifecycle event v2 parser")
}
sqsMessageParser, err := terminator.NewSqsMessageParser(event.NewParser(
sqsMessageParser, err := terminator.NewSQSMessageParser(event.NewParser(
asgTerminateEventV1Parser,
asgTerminateEventV2Parser,
rebalancerecommendationeventv0.NewParser(),
Expand All @@ -182,7 +182,7 @@ func main() {
if err != nil {
logger.With("error", err).Fatal("failed to create SQS message client")
}
terminatorSqsClientBuilder, err := terminator.NewSqsClientBuilder(sqsMessageClient)
terminatorSQSClientBuilder, err := terminator.NewSQSClientBuilder(sqsMessageClient)
if err != nil {
logger.With("error", err).Fatal("failed to create terminator SQS message client builder")
}
Expand All @@ -205,8 +205,8 @@ func main() {
RequeueInterval: time.Duration(10) * time.Second,
NodeGetter: nodeGetter,
NodeNameGetter: nodeNameGetter,
SqsClientBuilder: terminatorSqsClientBuilder,
SqsMessageParser: sqsMessageParser,
SQSClientBuilder: terminatorSQSClientBuilder,
SQSMessageParser: sqsMessageParser,
Getter: terminatorGetter,
CordonDrainerBuilder: terminatorCordonDrainerBuilder,
}
Expand Down Expand Up @@ -244,7 +244,7 @@ func parseOptions() Options {
return options
}

func newAwsSession() (*session.Session, error) {
func newAWSSession() (*session.Session, error) {
config := aws.NewConfig().WithSTSRegionalEndpoint(endpoints.RegionalSTSEndpoint)
sess, err := session.NewSessionWithOptions(session.Options{
Config: *config,
Expand Down
20 changes: 10 additions & 10 deletions src/pkg/event/asgterminate/v1/awsevent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,33 +23,33 @@ import (
"go.uber.org/zap/zapcore"
)

// AwsEvent contains the properties defined in AWS EventBridge schema
// AWSEvent contains the properties defined in AWS EventBridge schema
// aws.autoscaling@EC2InstanceTerminateLifecycleAction v1.
type AwsEvent struct {
event.AwsMetadata
type AWSEvent struct {
event.AWSMetadata

Detail Ec2InstanceTerminateLifecycleActionDetail `json:"detail"`
Detail EC2InstanceTerminateLifecycleActionDetail `json:"detail"`
}

func (e AwsEvent) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AwsMetadata).AddTo(enc)
func (e AWSEvent) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AWSMetadata).AddTo(enc)
enc.AddObject("detail", e.Detail)
return nil
}

type Ec2InstanceTerminateLifecycleActionDetail struct {
type EC2InstanceTerminateLifecycleActionDetail struct {
LifecycleHookName string `json:"LifecycleHookName"`
LifecycleTransition string `json:"LifecycleTransition"`
AutoScalingGroupName string `json:"AutoScalingGroupName"`
Ec2InstanceId string `json:"EC2InstanceId"`
EC2InstanceId string `json:"EC2InstanceId"`
LifecycleActionToken string `json:"LifecycleActionToken"`
}

func (e Ec2InstanceTerminateLifecycleActionDetail) MarshalLogObject(enc zapcore.ObjectEncoder) error {
func (e EC2InstanceTerminateLifecycleActionDetail) MarshalLogObject(enc zapcore.ObjectEncoder) error {
enc.AddString("LifecycleHookName", e.LifecycleHookName)
enc.AddString("LifecycleTransition", e.LifecycleTransition)
enc.AddString("AutoScalingGroupName", e.AutoScalingGroupName)
enc.AddString("EC2InstanceId", e.Ec2InstanceId)
enc.AddString("EC2InstanceId", e.EC2InstanceId)
enc.AddString("LifecycleActionToken", e.LifecycleActionToken)
return nil
}
18 changes: 9 additions & 9 deletions src/pkg/event/asgterminate/v1/event.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,22 +28,22 @@ import (
"go.uber.org/zap/zapcore"
)

type Ec2InstanceTerminateLifecycleAction struct {
AsgLifecycleActionCompleter
AwsEvent
type EC2InstanceTerminateLifecycleAction struct {
ASGLifecycleActionCompleter
AWSEvent
}

func (e Ec2InstanceTerminateLifecycleAction) Ec2InstanceIds() []string {
return []string{e.Detail.Ec2InstanceId}
func (e EC2InstanceTerminateLifecycleAction) EC2InstanceIds() []string {
return []string{e.Detail.EC2InstanceId}
}

func (e Ec2InstanceTerminateLifecycleAction) Done(ctx context.Context) (bool, error) {
func (e EC2InstanceTerminateLifecycleAction) Done(ctx context.Context) (bool, error) {
if _, err := e.CompleteLifecycleActionWithContext(ctx, &autoscaling.CompleteLifecycleActionInput{
AutoScalingGroupName: aws.String(e.Detail.AutoScalingGroupName),
LifecycleActionResult: aws.String("CONTINUE"),
LifecycleHookName: aws.String(e.Detail.LifecycleHookName),
LifecycleActionToken: aws.String(e.Detail.LifecycleActionToken),
InstanceId: aws.String(e.Detail.Ec2InstanceId),
InstanceId: aws.String(e.Detail.EC2InstanceId),
}); err != nil {
var f awserr.RequestFailure
return errors.As(err, &f) && f.StatusCode() != 400, err
Expand All @@ -52,7 +52,7 @@ func (e Ec2InstanceTerminateLifecycleAction) Done(ctx context.Context) (bool, er
return false, nil
}

func (e Ec2InstanceTerminateLifecycleAction) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AwsEvent).AddTo(enc)
func (e EC2InstanceTerminateLifecycleAction) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AWSEvent).AddTo(enc)
return nil
}
10 changes: 5 additions & 5 deletions src/pkg/event/asgterminate/v1/parser.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,21 +33,21 @@ const (
)

type parser struct {
AsgLifecycleActionCompleter
ASGLifecycleActionCompleter
}

func NewParser(completer AsgLifecycleActionCompleter) (event.Parser, error) {
func NewParser(completer ASGLifecycleActionCompleter) (event.Parser, error) {
if completer == nil {
return nil, fmt.Errorf("argument 'completer' is nil")
}
return parser{AsgLifecycleActionCompleter: completer}, nil
return parser{ASGLifecycleActionCompleter: completer}, nil
}

func (p parser) Parse(ctx context.Context, str string) event.Event {
ctx = logging.WithLogger(ctx, logging.FromContext(ctx).Named("asgTerminateLifecycleAction.v1"))

evt := Ec2InstanceTerminateLifecycleAction{
AsgLifecycleActionCompleter: p.AsgLifecycleActionCompleter,
evt := EC2InstanceTerminateLifecycleAction{
ASGLifecycleActionCompleter: p.ASGLifecycleActionCompleter,
}
if err := json.Unmarshal([]byte(str), &evt); err != nil {
logging.FromContext(ctx).
Expand Down
2 changes: 1 addition & 1 deletion src/pkg/event/asgterminate/v1/types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,6 @@ import (
"github.com/aws/aws-sdk-go/service/autoscaling"
)

type AsgLifecycleActionCompleter interface {
type ASGLifecycleActionCompleter interface {
CompleteLifecycleActionWithContext(aws.Context, *autoscaling.CompleteLifecycleActionInput, ...request.Option) (*autoscaling.CompleteLifecycleActionOutput, error)
}
20 changes: 10 additions & 10 deletions src/pkg/event/asgterminate/v2/awsevent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,36 +23,36 @@ import (
"go.uber.org/zap/zapcore"
)

// AwsEvent contains the properties defined in AWS EventBridge schema
// AWSEvent contains the properties defined in AWS EventBridge schema
// aws.autoscaling@EC2InstanceTerminateLifecycleAction v2.
type AwsEvent struct {
event.AwsMetadata
type AWSEvent struct {
event.AWSMetadata

Detail Ec2InstanceTerminateLifecycleActionDetail `json:"detail"`
Detail EC2InstanceTerminateLifecycleActionDetail `json:"detail"`
}

func (e AwsEvent) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AwsMetadata).AddTo(enc)
func (e AWSEvent) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AWSMetadata).AddTo(enc)
enc.AddObject("detail", e.Detail)
return nil
}

type Ec2InstanceTerminateLifecycleActionDetail struct {
type EC2InstanceTerminateLifecycleActionDetail struct {
LifecycleHookName string `json:"LifecycleHookName"`
LifecycleTransition string `json:"LifecycleTransition"`
AutoScalingGroupName string `json:"AutoScalingGroupName"`
Ec2InstanceId string `json:"EC2InstanceId"`
EC2InstanceId string `json:"EC2InstanceId"`
LifecycleActionToken string `json:"LifecycleActionToken"`
NotificationMetadata string `json:"NotificationMetadata"`
Origin string `json:"Origin"`
Destination string `json:"Destination"`
}

func (e Ec2InstanceTerminateLifecycleActionDetail) MarshalLogObject(enc zapcore.ObjectEncoder) error {
func (e EC2InstanceTerminateLifecycleActionDetail) MarshalLogObject(enc zapcore.ObjectEncoder) error {
enc.AddString("LifecycleHookName", e.LifecycleHookName)
enc.AddString("LifecycleTransition", e.LifecycleTransition)
enc.AddString("AutoScalingGroupName", e.AutoScalingGroupName)
enc.AddString("EC2InstanceId", e.Ec2InstanceId)
enc.AddString("EC2InstanceId", e.EC2InstanceId)
enc.AddString("LifecycleActionToken", e.LifecycleActionToken)
return nil
}
18 changes: 9 additions & 9 deletions src/pkg/event/asgterminate/v2/event.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,22 +28,22 @@ import (
"go.uber.org/zap/zapcore"
)

type Ec2InstanceTerminateLifecycleAction struct {
AsgLifecycleActionCompleter
AwsEvent
type EC2InstanceTerminateLifecycleAction struct {
ASGLifecycleActionCompleter
AWSEvent
}

func (e Ec2InstanceTerminateLifecycleAction) Ec2InstanceIds() []string {
return []string{e.Detail.Ec2InstanceId}
func (e EC2InstanceTerminateLifecycleAction) EC2InstanceIds() []string {
return []string{e.Detail.EC2InstanceId}
}

func (e Ec2InstanceTerminateLifecycleAction) Done(ctx context.Context) (bool, error) {
func (e EC2InstanceTerminateLifecycleAction) Done(ctx context.Context) (bool, error) {
if _, err := e.CompleteLifecycleActionWithContext(ctx, &autoscaling.CompleteLifecycleActionInput{
AutoScalingGroupName: aws.String(e.Detail.AutoScalingGroupName),
LifecycleActionResult: aws.String("CONTINUE"),
LifecycleHookName: aws.String(e.Detail.LifecycleHookName),
LifecycleActionToken: aws.String(e.Detail.LifecycleActionToken),
InstanceId: aws.String(e.Detail.Ec2InstanceId),
InstanceId: aws.String(e.Detail.EC2InstanceId),
}); err != nil {
var f awserr.RequestFailure
return errors.As(err, &f) && f.StatusCode() != 400, err
Expand All @@ -52,7 +52,7 @@ func (e Ec2InstanceTerminateLifecycleAction) Done(ctx context.Context) (bool, er
return false, nil
}

func (e Ec2InstanceTerminateLifecycleAction) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AwsEvent).AddTo(enc)
func (e EC2InstanceTerminateLifecycleAction) MarshalLogObject(enc zapcore.ObjectEncoder) error {
zap.Inline(e.AWSEvent).AddTo(enc)
return nil
}
Loading