feat: ability to emit Kubernetes events

README.md

@@ -31,11 +31,11 @@
 
 ## Project Summary
 
-This project ensures that the Kubernetes control plane responds appropriately to events that can cause your EC2 instance to become unavailable, such as [EC2 maintenance events](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/monitoring-instances-status-check_sched.html), [EC2 Spot interruptions](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/spot-interruptions.html), [ASG Scale-In](https://docs.aws.amazon.com/autoscaling/ec2/userguide/AutoScalingGroupLifecycle.html#as-lifecycle-scale-in), [ASG AZ Rebalance](https://docs.aws.amazon.com/autoscaling/ec2/userguide/auto-scaling-benefits.html#AutoScalingBehavior.InstanceUsage), and EC2 Instance Termination via the API or Console.  If not handled, your application code may not stop gracefully, take longer to recover full availability, or accidentally schedule work to nodes that are going down. 
+This project ensures that the Kubernetes control plane responds appropriately to events that can cause your EC2 instance to become unavailable, such as [EC2 maintenance events](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/monitoring-instances-status-check_sched.html), [EC2 Spot interruptions](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/spot-interruptions.html), [ASG Scale-In](https://docs.aws.amazon.com/autoscaling/ec2/userguide/AutoScalingGroupLifecycle.html#as-lifecycle-scale-in), [ASG AZ Rebalance](https://docs.aws.amazon.com/autoscaling/ec2/userguide/auto-scaling-benefits.html#AutoScalingBehavior.InstanceUsage), and EC2 Instance Termination via the API or Console.  If not handled, your application code may not stop gracefully, take longer to recover full availability, or accidentally schedule work to nodes that are going down.
 
-The aws-node-termination-handler (NTH) can operate in two different modes: Instance Metadata Service (IMDS) or the Queue Processor. 
+The aws-node-termination-handler (NTH) can operate in two different modes: Instance Metadata Service (IMDS) or the Queue Processor.
 
-The aws-node-termination-handler **[Instance Metadata Service](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html) Monitor** will run a small pod on each host to perform monitoring of IMDS paths like `/spot` or `/events` and react accordingly to drain and/or cordon the corresponding node. 
+The aws-node-termination-handler **[Instance Metadata Service](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html) Monitor** will run a small pod on each host to perform monitoring of IMDS paths like `/spot` or `/events` and react accordingly to drain and/or cordon the corresponding node.
 
 The aws-node-termination-handler **Queue Processor** will monitor an SQS queue of events from Amazon EventBridge for ASG lifecycle events, EC2 status change events, and Spot Interruption Termination Notice events. When NTH detects an instance is going down, we use the Kubernetes API to cordon the node to ensure no new work is scheduled there, then drain it, removing any existing work. The termination handler **Queue Processor** requires AWS IAM permissions to monitor and manage the SQS queue and to query the EC2 API. The queue processor mode is currently in a beta preview, but we'd love your feedback on it!
 
@@ -52,7 +52,7 @@ You can run the termination handler on any Kubernetes cluster running on AWS, in
 - Unit & Integration Tests
 
 ### Queue Processor
-- Monitors an SQS Queue for: 
+- Monitors an SQS Queue for:
    - EC2 Spot Interruption Notifications
    - EC2 Instance Rebalance Recommendation
    - EC2 Auto-Scaling Group Termination Lifecycle Hooks to take care of ASG Scale-In, AZ-Rebalance, Unhealthy Instances, and more!
@@ -82,10 +82,10 @@ IMDS Processor Mode allows for a fine-grained configuration of IMDS paths that a
  - `enableSpotInterruptionDraining`
  - `enableRebalanceMonitoring`
  - `enableScheduledEventDraining`
- 
+
 The `enableSqsTerminationDraining` must be set to false for these configuration values to be considered.
 
-The Queue Processor Mode does not allow for fine-grained configuration of which events are handled through helm configuration keys. Instead, you can modify your Amazon EventBridge rules to not send certain types of events to the SQS Queue so that NTH does not process those events. 
+The Queue Processor Mode does not allow for fine-grained configuration of which events are handled through helm configuration keys. Instead, you can modify your Amazon EventBridge rules to not send certain types of events to the SQS Queue so that NTH does not process those events.
 
 
 The `enableSqsTerminationDraining` flag turns on Queue Processor Mode. When Queue Processor Mode is enabled, IMDS mode cannot be active. NTH cannot respond to queue events AND monitor IMDS paths. Queue Processor Mode still queries for node information on startup, but this information is not required for normal operation, so it is safe to disable IMDS for the NTH pod.
@@ -102,6 +102,7 @@ The termination handler DaemonSet installs into your cluster a [ServiceAccount](
 #### Kubectl Apply
 
 You can use kubectl to directly add all of the above resources with the default configuration into your cluster.
+
 ```
 kubectl apply -f https://github.com/aws/aws-node-termination-handler/releases/download/v1.13.0/all-resources.yaml
 ```
@@ -121,13 +122,15 @@ helm repo add eks https://aws.github.io/eks-charts
 Once that is complete you can install the termination handler. We've provided some sample setup options below.
 
 Zero Config:
+
 ```sh
 helm upgrade --install aws-node-termination-handler \
   --namespace kube-system \
   eks/aws-node-termination-handler
 ```
 
 Enabling Features:
+
 ```
 helm upgrade --install aws-node-termination-handler \
   --namespace kube-system \
@@ -140,6 +143,7 @@ helm upgrade --install aws-node-termination-handler \
 The `enable*` configuration flags above enable or disable IMDS monitoring paths.
 
 Running Only On Specific Nodes:
+
 ```
 helm upgrade --install aws-node-termination-handler \
   --namespace kube-system \
@@ -148,6 +152,7 @@ helm upgrade --install aws-node-termination-handler \
 ```
 
 Webhook Configuration:
+
 ```
 helm upgrade --install aws-node-termination-handler \
   --namespace kube-system \
@@ -156,6 +161,7 @@ helm upgrade --install aws-node-termination-handler \
 ```
 
 Alternatively, pass Webhook URL as a Secret:
+
 ```
 WEBHOOKURL_LITERAL="webhookurl=https://hooks.slack.com/services/YOUR/SLACK/URL"
 
@@ -212,16 +218,14 @@ $ aws autoscaling create-or-update-tags \
 
 The value of the key does not matter.
 
-This functionality is helpful in accounts where there are ASGs that do not run kubernetes nodes or you do not want aws-node-termination-handler to manage their termination lifecycle. 
+This functionality is helpful in accounts where there are ASGs that do not run kubernetes nodes or you do not want aws-node-termination-handler to manage their termination lifecycle.
 However, if your account is dedicated to ASGs for your kubernetes cluster, then you can turn off the ASG tag check by setting the flag `--check-asg-tag-before-draining=false` or environment variable `CHECK_ASG_TAG_BEFORE_DRAINING=false`.
 
-You can also control what resources NTH manages by adding the resource ARNs to your Amazon EventBridge rules. 
-
-Take a look at the docs on how to create rules that only manage certain ASGs here: https://docs.aws.amazon.com/autoscaling/ec2/userguide/cloud-watch-events.html 
-
-See all the different events docs here: https://docs.aws.amazon.com/eventbridge/latest/userguide/event-types.html#auto-scaling-event-types
+You can also control what resources NTH manages by adding the resource ARNs to your Amazon EventBridge rules.
 
+Take a look at the docs on how to create rules that only manage certain ASGs [here](https://docs.aws.amazon.com/autoscaling/ec2/userguide/cloud-watch-events.html).
 
+See all the different events docs [here](https://docs.aws.amazon.com/eventbridge/latest/userguide/event-types.html#auto-scaling-event-types).
 
 #### 3. Create an SQS Queue:
 
@@ -233,7 +237,7 @@ $ QUEUE_POLICY=$(cat <<EOF
 {
     "Version": "2012-10-17",
     "Id": "MyQueuePolicy",
-    "Statement": [{                     
+    "Statement": [{
         "Effect": "Allow",
         "Principal": {
             "Service": ["events.amazonaws.com", "sqs.amazonaws.com"]
@@ -248,17 +252,17 @@ EOF
 )
 
 ## make sure the queue policy is valid JSON
-$ echo "$QUEUE_POLICY" | jq . 
+$ echo "$QUEUE_POLICY" | jq .
 
-## Save queue attributes to a temp file 
+## Save queue attributes to a temp file
 $ cat << EOF > /tmp/queue-attributes.json
 {
   "MessageRetentionPeriod": "300",
   "Policy": "$(echo $QUEUE_POLICY | sed 's/\"/\\"/g')"
 }
 EOF
 
-$ aws sqs create-queue --queue-name "${SQS_QUEUE_NAME}" --attributes file:///tmp/queue-attributes.json 
+$ aws sqs create-queue --queue-name "${SQS_QUEUE_NAME}" --attributes file:///tmp/queue-attributes.json
 ```
 
 #### 4. Create an Amazon EventBridge Rule
@@ -298,6 +302,7 @@ There are many different ways to allow the aws-node-termination-handler pods to
 4. [kube2iam](https://github.com/jtblin/kube2iam)
 
 IAM Policy for aws-node-termination-handler Deployment:
+
 ```
 {
     "Version": "2012-10-17",
@@ -333,6 +338,7 @@ helm repo add eks https://aws.github.io/eks-charts
 Once that is complete you can install the termination handler. We've provided some sample setup options below.
 
 Minimal Config:
+
 ```sh
 helm upgrade --install aws-node-termination-handler \
   --namespace kube-system \
@@ -342,6 +348,7 @@ helm upgrade --install aws-node-termination-handler \
 ```
 
 Webhook Configuration:
+
 ```
 helm upgrade --install aws-node-termination-handler \
   --namespace kube-system \
@@ -352,6 +359,7 @@ helm upgrade --install aws-node-termination-handler \
 ```
 
 Alternatively, pass Webhook URL as a Secret:
+
 ```
 WEBHOOKURL_LITERAL="webhookurl=https://hooks.slack.com/services/YOUR/SLACK/URL"
 
@@ -389,37 +397,42 @@ For a full list of releases and associated artifacts see our [releases page](htt
 <summary>Use with Kiam</summary>
 <br>
 
-## Use with Kiam 
+## Use with Kiam
 
 If you are using IMDS mode which defaults to `hostNetworking: true`, or if you are using queue-processor mode, then this section does not apply. The configuration below only needs to be used if you are explicitly changing NTH IMDS mode to `hostNetworking: false` .
 
 To use the termination handler alongside [Kiam](https://github.com/uswitch/kiam) requires some extra configuration on Kiam's end.
 By default Kiam will block all access to the metadata address, so you need to make sure it passes through the requests the termination handler relies on.
 
 To add a whitelist configuration, use the following fields in the Kiam Helm chart values:
+
 ```
 agent.whiteListRouteRegexp: '^\/latest\/meta-data\/(spot\/instance-action|events\/maintenance\/scheduled|instance-(id|type)|public-(hostname|ipv4)|local-(hostname|ipv4)|placement\/availability-zone)|\/latest\/dynamic\/instance-identity\/document$'
 ```
 Or just pass it as an argument to the kiam agents:
+
 ```
 kiam agent --whitelist-route-regexp='^\/latest\/meta-data\/(spot\/instance-action|events\/maintenance\/scheduled|instance-(id|type)|public-(hostname|ipv4)|local-(hostname|ipv4)|placement\/availability-zone)|\/latest\/dynamic\/instance-identity\/document$'
 ```
 
 ## Metadata endpoints
 The termination handler relies on the following metadata endpoints to function properly:
+
 ```
 /latest/dynamic/instance-identity/document
 /latest/meta-data/spot/instance-action
 /latest/meta-data/events/recommendations/rebalance
 /latest/meta-data/events/maintenance/scheduled
 /latest/meta-data/instance-id
+/latest/meta-data/instance-life-cycle
 /latest/meta-data/instance-type
 /latest/meta-data/public-hostname
 /latest/meta-data/public-ipv4
 /latest/meta-data/local-hostname
 /latest/meta-data/local-ipv4
 /latest/meta-data/placement/availability-zone
 ```
+
 </details>
 
 ## Building
@@ -428,7 +441,7 @@ For build instructions please consult [BUILD.md](./BUILD.md).
 ## Communication
 * If you've run into a bug or have a new feature request, please open an [issue](https://github.com/aws/aws-node-termination-handler/issues/new).
 * You can also chat with us in the [Kubernetes Slack](https://kubernetes.slack.com) in the `#provider-aws` channel
-* Check out the open source [Amazon EC2 Spot Instances Integrations Roadmap](https://github.com/aws/ec2-spot-instances-integrations-roadmap) to see what we're working on and give us feedback! 
+* Check out the open source [Amazon EC2 Spot Instances Integrations Roadmap](https://github.com/aws/ec2-spot-instances-integrations-roadmap) to see what we're working on and give us feedback!
 
 ##  Contributing
 Contributions are welcome! Please read our [guidelines](https://github.com/aws/aws-node-termination-handler/blob/main/CONTRIBUTING.md) and our [Code of Conduct](https://github.com/aws/aws-node-termination-handler/blob/main/CODE_OF_CONDUCT.md)

cmd/node-termination-handler.go

@@ -113,6 +113,13 @@ func main() {
 		nthConfig.Print()
 		log.Fatal().Msgf("Unable to find the AWS region to process queue events.")
 	}
+
+	recorder, err := observability.InitK8sEventRecorder(nthConfig.EmitKubernetesEvents, nthConfig.NodeName, nodeMetadata, nthConfig.KubernetesEventsExtraAnnotations)
+	if err != nil {
+		nthConfig.Print()
+		log.Fatal().Err(err).Msg("Unable to create Kubernetes event recorder,")
+	}
+
 	nthConfig.Print()
 
 	if nthConfig.EnableScheduledEventDraining {
@@ -179,6 +186,7 @@ func main() {
 				if err != nil {
 					log.Warn().Str("event_type", monitor.Kind()).Err(err).Msg("There was a problem monitoring for events")
 					metrics.ErrorEventsInc(monitor.Kind())
+					recorder.Emit(nthConfig.NodeName, observability.Warning, observability.MonitorErrReason, observability.MonitorErrMsgFmt, monitor.Kind())
 					if previousErr != nil && err.Error() == previousErr.Error() {
 						duplicateErrCount++
 					} else {
@@ -198,7 +206,7 @@ func main() {
 	log.Info().Msg("Started watching for interruption events")
 	log.Info().Msg("Kubernetes AWS Node Termination Handler has started successfully!")
 
-	go watchForCancellationEvents(cancelChan, interruptionEventStore, node, metrics)
+	go watchForCancellationEvents(cancelChan, interruptionEventStore, node, metrics, recorder)
 	log.Info().Msg("Started watching for event cancellations")
 
 	var wg sync.WaitGroup
@@ -214,7 +222,8 @@ func main() {
 				case interruptionEventStore.Workers <- 1:
 					event.InProgress = true
 					wg.Add(1)
-					go drainOrCordonIfNecessary(interruptionEventStore, event, *node, nthConfig, nodeMetadata, metrics, &wg)
+					recorder.Emit(event.NodeName, observability.Normal, observability.GetReasonForKind(event.Kind), event.Description)
+					go drainOrCordonIfNecessary(interruptionEventStore, event, *node, nthConfig, nodeMetadata, metrics, recorder, &wg)
 				default:
 					log.Warn().Msg("all workers busy, waiting")
 					break
@@ -254,7 +263,7 @@ func watchForInterruptionEvents(interruptionChan <-chan monitor.InterruptionEven
 	}
 }
 
-func watchForCancellationEvents(cancelChan <-chan monitor.InterruptionEvent, interruptionEventStore *interruptioneventstore.Store, node *node.Node, metrics observability.Metrics) {
+func watchForCancellationEvents(cancelChan <-chan monitor.InterruptionEvent, interruptionEventStore *interruptioneventstore.Store, node *node.Node, metrics observability.Metrics, recorder observability.K8sEventRecorder) {
 	for {
 		interruptionEvent := <-cancelChan
 		nodeName := interruptionEvent.NodeName
@@ -264,6 +273,9 @@ func watchForCancellationEvents(cancelChan <-chan monitor.InterruptionEvent, int
 			err := node.Uncordon(nodeName)
 			if err != nil {
 				log.Err(err).Msg("Uncordoning the node failed")
+				recorder.Emit(nodeName, observability.Warning, observability.UncordonErrReason, observability.UncordonErrMsgFmt, err.Error())
+			} else {
+				recorder.Emit(nodeName, observability.Normal, observability.UncordonReason, observability.UncordonMsg)
 			}
 			metrics.NodeActionsInc("uncordon", nodeName, err)
 
@@ -275,7 +287,7 @@ func watchForCancellationEvents(cancelChan <-chan monitor.InterruptionEvent, int
 	}
 }
 
-func drainOrCordonIfNecessary(interruptionEventStore *interruptioneventstore.Store, drainEvent *monitor.InterruptionEvent, node node.Node, nthConfig config.Config, nodeMetadata ec2metadata.NodeMetadata, metrics observability.Metrics, wg *sync.WaitGroup) {
+func drainOrCordonIfNecessary(interruptionEventStore *interruptioneventstore.Store, drainEvent *monitor.InterruptionEvent, node node.Node, nthConfig config.Config, nodeMetadata ec2metadata.NodeMetadata, metrics observability.Metrics, recorder observability.K8sEventRecorder, wg *sync.WaitGroup) {
 	defer wg.Done()
 	nodeName := drainEvent.NodeName
 	nodeLabels, err := node.GetNodeLabels(nodeName)
@@ -287,6 +299,9 @@ func drainOrCordonIfNecessary(interruptionEventStore *interruptioneventstore.Sto
 		err := drainEvent.PreDrainTask(*drainEvent, node)
 		if err != nil {
 			log.Err(err).Msg("There was a problem executing the pre-drain task")
+			recorder.Emit(nodeName, observability.Warning, observability.PreDrainErrReason, observability.PreDrainErrMsgFmt, err.Error())
+		} else {
+			recorder.Emit(nodeName, observability.Normal, observability.PreDrainReason, observability.PreDrainMsg)
 		}
 		metrics.NodeActionsInc("pre-drain", nodeName, err)
 	}
@@ -298,6 +313,7 @@ func drainOrCordonIfNecessary(interruptionEventStore *interruptioneventstore.Sto
 				log.Err(err).Msgf("node '%s' not found in the cluster", nodeName)
 			} else {
 				log.Err(err).Msg("There was a problem while trying to cordon the node")
+				recorder.Emit(nodeName, observability.Warning, observability.CordonErrReason, observability.CordonErrMsgFmt, err.Error())
 				os.Exit(1)
 			}
 		} else {
@@ -312,6 +328,7 @@ func drainOrCordonIfNecessary(interruptionEventStore *interruptioneventstore.Sto
 				log.Err(err).Msg("There was a problem while trying to log all pod names on the node")
 			}
 			metrics.NodeActionsInc("cordon", nodeName, err)
+			recorder.Emit(nodeName, observability.Normal, observability.CordonReason, observability.CordonMsg)
 		}
 	} else {
 		err := node.CordonAndDrain(nodeName)
@@ -320,11 +337,14 @@ func drainOrCordonIfNecessary(interruptionEventStore *interruptioneventstore.Sto
 				log.Err(err).Msgf("node '%s' not found in the cluster", nodeName)
 			} else {
 				log.Err(err).Msg("There was a problem while trying to cordon and drain the node")
+				metrics.NodeActionsInc("cordon-and-drain", nodeName, err)
+				recorder.Emit(nodeName, observability.Warning, observability.CordonAndDrainErrReason, observability.CordonAndDrainErrMsgFmt, err.Error())
 				os.Exit(1)
 			}
 		} else {
 			log.Info().Str("node_name", nodeName).Msg("Node successfully cordoned and drained")
 			metrics.NodeActionsInc("cordon-and-drain", nodeName, err)
+			recorder.Emit(nodeName, observability.Normal, observability.CordonAndDrainReason, observability.CordonAndDrainMsg)
 		}
 	}
 
@@ -336,6 +356,9 @@ func drainOrCordonIfNecessary(interruptionEventStore *interruptioneventstore.Sto
 		err := drainEvent.PostDrainTask(*drainEvent, node)
 		if err != nil {
 			log.Err(err).Msg("There was a problem executing the post-drain task")
+			recorder.Emit(nodeName, observability.Warning, observability.PostDrainErrReason, observability.PostDrainErrMsgFmt, err.Error())
+		} else {
+			recorder.Emit(nodeName, observability.Normal, observability.PostDrainReason, observability.PostDrainMsg)
 		}
 		metrics.NodeActionsInc("post-drain", nodeName, err)
 	}