core functionality

* Add SQS parameters to Terminator spec
* Add drain parameters to Terminator spec
* Fetch messages from SQS queue
* Cordon and drain nodes targeted in SQS messages

BUILD.md

@@ -1,30 +1,78 @@
 # Setup Development Environment
 
-Clone the repo:
+## Clone the repo
 
 ```sh
 git clone --branch v2 https://github.com/aws/aws-node-termination-handler.git 
+cd aws-node-termination-handler
 ```
 
-Install build tools
+## Install build tools
 
 ```sh
 make toolchain
 ```
 
-Configure image repository location
+## Set environment variables
 
 ```sh
-export KO_DOCKER_REPO=my.image.repo/path
+export AWS_REGION=<region>
+export AWS_ACCOUNT_ID=<account-id>
+export CLUSTER_NAME=<name>
 ```
 
-Build and deploy controller to Kubernetes cluster
+## Create Image Repositories
+
+```sh
+aws ecr create-repository \
+    --repository-name nthv2/controller \
+    --image-scanning-configuration scanOnPush=true \
+    --region "${AWS_REGION}"
+
+aws ecr create-repository \
+    --repository-name nthv2/webhook \
+    --image-scanning-configuration scanOnPush=true \
+    --region "${AWS_REGION}"
+
+export KO_DOCKER_REPO="${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com/nthv2"
+
+./scripts/docker-login-ecr.sh
+```
+
+## Create an EKS Cluster
+
+```sh
+envsubst src/resources/eks-cluster.yaml.tmpl | eksctl create cluster -f -
+```
+
+### Create the Controller IAM Role
+
+```sh
+aws cloudformation deploy \
+    --template-file src/resources/controller-iam-role.yaml \
+    --stack-name "nthv2-${CLUSTER_NAME}" \
+    --capabilities CAPABILITY_NAMED_IAM \
+    --parameter-overrides "ClusterName=${CLUSTER_NAME}"
+
+eksctl create iamserviceaccount \
+    --cluster "${CLUSTER_NAME}" \
+    --name nthv2 \
+    --namespace nthv2 \
+    --role-name "${CLUSTER_NAME}-nthv2" \
+    --attach-policy-arn "arn:aws:iam::${AWS_ACCOUNT_ID}:policy/Nthv2ControllerPolicy-${CLUSTER_NAME}" \
+    --role-only \
+    --approve
+
+export NTHV2_IAM_ROLE_ARN="arn:aws:iam::${AWS_ACCOUNT_ID}:role/${CLUSTER_NAME}-nthv2
+```
+
+## Build and deploy controller to Kubernetes cluster
 
 ```sh
 make apply
 ```
 
-Remove deployed controller from Kubernetes cluster
+## Remove deployed controller from Kubernetes cluster
 
 ```sh
 make delete

src/Makefile

@@ -2,6 +2,8 @@ PROJECT_DIR := $(shell dirname $(abspath $(lastword $(MAKEFILE_LIST))))
 CONTROLLER_GEN = $(PROJECT_DIR)/bin/controller-gen
 KO = $(PROJECT_DIR)/bin/ko
 ENVTEST = $(PROJECT_DIR)/bin/setup-envtest
+HELM_BASE_OPTS ?= --set serviceAccount.annotations.eks\\.amazonaws\\.com/role-arn=${NTHV2_IAM_ROLE_ARN}
+GINKGO_BASE_OPTS ?= --coverpkg $(shell head -n 1 $(PROJECT_DIR)/go.mod | cut -s -d ' ' -f 2)/pkg/...
 
 
 # Image URL to use all building/pushing image targets
@@ -45,34 +47,32 @@ toolchain:  ## Download additional tools.
 
 ##@ Development
 
-.PHONY: manifests
-manifests:  ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects.
-	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="./..." output:crd:artifacts:config=config/crd/bases
-
 .PHONY: generate
 generate:  ## Generate code containing DeepCopy, DeepCopyInto, and DeepCopyObject method implementations.
 	$(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./..."
 
 .PHONY: verify
 verify: ## Run go fmt and go vet against code.
-	go fmt ./...
-	go vet ./...
+	go fmt $(PROJECT_DIR)/...
+	go vet $(PROJECT_DIR)/...
 
 .PHONY: test
-test: manifests generate fmt vet envtest ## Run tests.
-	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) -p path)" go test ./... -coverprofile cover.out
+test: ## Run tests.
+	go vet $(PROJECT_DIR)/...
+	ginkgo run $(GINKGO_BASE_OPTS) $(GINKGO_OPTS) $(PROJECT_DIR)/test/
 
 ##@ Build
 
 .PHONY: run
-run: manifests generate fmt vet ## Run a controller from your host.
+run: ## Run a controller from your host.
 	go run ./main.go
 
 ##@ Deployment
 
 .PHONY: apply
 apply:  ## Deploy the controller into the current kubernetes cluster.
 	helm upgrade --install dev charts/aws-node-termination-handler-2 --namespace nthv2 --create-namespace \
+		$(HELM_BASE_OPTS) \
 		$(HELM_OPTS) \
 		--set controller.image=$(shell $(KO) publish -B github.com/aws/aws-node-termination-handler/cmd/controller) \
 		--set webhook.image=$(shell $(KO) publish -B github.com/aws/aws-node-termination-handler/cmd/webhook) \

src/api/v1alpha1/terminator_logging.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2022.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+import (
+	"go.uber.org/zap/zapcore"
+)
+
+func (t *TerminatorSpec) MarshalLogObject(enc zapcore.ObjectEncoder) error {
+	enc.AddObject("sqs", t.Sqs)
+	enc.AddObject("drain", t.Drain)
+	return nil
+}
+
+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)
+		}
+		return nil
+	}))
+
+	enc.AddInt64("maxNumberOfMessages", s.MaxNumberOfMessages)
+
+	enc.AddArray("messageAttributeNames", zapcore.ArrayMarshalerFunc(func(enc zapcore.ArrayEncoder) error {
+		for _, attrName := range s.MessageAttributeNames {
+			enc.AppendString(attrName)
+		}
+		return nil
+	}))
+
+	enc.AddString("queueUrl", s.QueueUrl)
+	enc.AddInt64("visibilityTimeoutSeconds", s.VisibilityTimeoutSeconds)
+	enc.AddInt64("waitTimeSeconds", s.WaitTimeSeconds)
+	return nil
+}
+
+func (d DrainSpec) MarshalLogObject(enc zapcore.ObjectEncoder) error {
+	enc.AddBool("force", d.Force)
+	enc.AddInt("gracePeriodSeconds", d.GracePeriodSeconds)
+	enc.AddBool("ignoreAllDaemonSets", d.IgnoreAllDaemonSets)
+	enc.AddBool("deleteEmptyDirData", d.DeleteEmptyDirData)
+	enc.AddInt("timeoutSeconds", d.TimeoutSeconds)
+	return nil
+}

src/api/v1alpha1/terminator_types.go

@@ -20,7 +20,6 @@ import (
 	"context"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/pkg/apis"
 )
 
 // EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
@@ -31,8 +30,29 @@ type TerminatorSpec struct {
 	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
 	// Important: Run "make" to regenerate code after modifying this file
 
-	// Foo is an example field of Terminator. Edit terminator_types.go to remove/update
-	Foo string `json:"foo,omitempty"`
+	Sqs   SqsSpec   `json:"sqs,omitempty"`
+	Drain DrainSpec `json:"drain,omitempty"`
+}
+
+// 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"`
+	VisibilityTimeoutSeconds int64    `json:"visibilityTimeoutSeconds,omitempty"`
+	WaitTimeSeconds          int64    `json:"waitTimeSeconds,omitempty"`
+}
+
+// DrainSpec defines inputs to the cordon and drain operations.
+type DrainSpec struct {
+	// https://pkg.go.dev/k8s.io/[email protected]/pkg/drain#Helper
+	Force               bool `json:"force,omitempty"`
+	GracePeriodSeconds  int  `json:"gracePeriodSeconds,omitempty"`
+	IgnoreAllDaemonSets bool `json:"ignoreAllDaemonSets,omitempty"`
+	DeleteEmptyDirData  bool `json:"deleteEmptyDirData,omitempty"`
+	TimeoutSeconds      int  `json:"timeoutSeconds,omitempty"`
 }
 
 // TerminatorStatus defines the observed state of Terminator
@@ -58,12 +78,6 @@ func (t *Terminator) SetDefaults(_ context.Context) {
 	// TODO: actually set defaults
 }
 
-func (t *Terminator) Validate(_ context.Context) *apis.FieldError {
-	// Stubbed to satisfy interface requirements.
-	// TODO: actually validate
-	return nil
-}
-
 // TerminatorList contains a list of Terminator
 //+kubebuilder:object:root=true
 type TerminatorList struct {

src/api/v1alpha1/terminator_validation.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2022.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+import (
+	"context"
+	"net/url"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/service/sqs"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	"knative.dev/pkg/apis"
+)
+
+var (
+	// https://github.com/aws/aws-sdk-go/blob/v1.38.55/service/sqs/api.go#L3966-L3994
+	knownSqsAttributeNames = sets.NewString(sqs.MessageSystemAttributeName_Values()...)
+)
+
+func (t *Terminator) Validate(_ context.Context) (errs *apis.FieldError) {
+	return errs.Also(
+		apis.ValidateObjectMetadata(t).ViaField("metadata"),
+		t.Spec.validate().ViaField("spec"),
+	)
+}
+
+func (t *TerminatorSpec) validate() (errs *apis.FieldError) {
+	return t.Sqs.validate().ViaField("sqs")
+}
+
+func (s *SqsSpec) validate() (errs *apis.FieldError) {
+	for _, attrName := range s.AttributeNames {
+		if !knownSqsAttributeNames.Has(attrName) {
+			errs = errs.Also(apis.ErrInvalidValue(attrName, "attributeNames"))
+		}
+	}
+
+	// https://github.com/aws/aws-sdk-go/blob/v1.38.55/service/sqs/api.go#L3996-L3999
+	if s.MaxNumberOfMessages < 1 || 10 < s.MaxNumberOfMessages {
+		errs = errs.Also(apis.ErrInvalidValue(s.MaxNumberOfMessages, "maxNumberOfMessages", "must be in range 1-10"))
+	}
+
+	// https://github.com/aws/aws-sdk-go/blob/v1.38.55/service/sqs/api.go#L4001-L4021
+	//
+	// Simple checks are done below. More indepth checks are left to the SQS client/service.
+	for _, attrName := range s.MessageAttributeNames {
+		if len(attrName) > 256 {
+			errs = errs.Also(apis.ErrInvalidValue(attrName, "messageAttributeNames", "must be 256 characters or less"))
+		}
+
+		lcAttrName := strings.ToLower(attrName)
+		if strings.HasPrefix(lcAttrName, "aws") || strings.HasPrefix(lcAttrName, "amazon") {
+			errs = errs.Also(apis.ErrInvalidValue(attrName, "messageAttributeNames", `must not use reserved prefixes "AWS" or "Amazon"`))
+		}
+
+		if strings.HasPrefix(attrName, ".") || strings.HasSuffix(attrName, ".") {
+			errs = errs.Also(apis.ErrInvalidValue(attrName, "messageAttributeNames", "must not begin or end with a period (.)"))
+		}
+	}
+
+	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 {
+		errs = errs.Also(apis.ErrInvalidValue(s.VisibilityTimeoutSeconds, "visibilityTimeoutSeconds", "must be zero or greater"))
+	}
+
+	if s.WaitTimeSeconds < 0 {
+		errs = errs.Also(apis.ErrInvalidValue(s.WaitTimeSeconds, "waitTimeSeconds", "must be zero or greater"))
+	}
+
+	return errs
+}