update reconciler

src/cmd/controller/main.go

@@ -57,6 +57,7 @@ import (
 	"github.com/aws/aws-node-termination-handler/pkg/sqsmessage"
 	"github.com/aws/aws-node-termination-handler/pkg/terminator"
 	terminatoradapter "github.com/aws/aws-node-termination-handler/pkg/terminator/adapter"
+	"github.com/aws/aws-node-termination-handler/pkg/webhook"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/ec2metadata"
@@ -172,6 +173,9 @@ func main() {
 				Drainer:   kubectlcordondrainer.DefaultDrainer,
 			},
 		},
+		WebhookClientBuilder: terminatoradapter.WebhookClientBuilder(
+			webhook.ClientBuilder(webhook.NewHttpClientDo).NewClient,
+		),
 	}
 	if err = rec.BuildController(
 		ctrl.NewControllerManagedBy(mgr).

src/pkg/terminator/adapter/webhookclient.go

@@ -0,0 +1,41 @@
+/*
+Copyright 2022 Amazon.com, Inc. or its affiliates. All rights reserved.
+
+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 adapter
+
+import (
+	"net/http"
+
+	"github.com/aws/aws-node-termination-handler/api/v1alpha1"
+	"github.com/aws/aws-node-termination-handler/pkg/terminator"
+	"github.com/aws/aws-node-termination-handler/pkg/webhook"
+)
+
+type WebhookClientBuilder func(url, proxyURL, template string, headers http.Header) (webhook.Client, error)
+
+func (b WebhookClientBuilder) NewWebhookClient(terminator *v1alpha1.Terminator) (terminator.WebhookClient, error) {
+	headers := http.Header{}
+	for _, h := range terminator.Spec.Webhook.Headers {
+		headers.Add(h.Name, h.Value)
+	}
+
+	return b(
+		terminator.Spec.Webhook.URL,
+		terminator.Spec.Webhook.ProxyURL,
+		terminator.Spec.Webhook.Template,
+		headers,
+	)
+}

src/pkg/terminator/eventkind.go

@@ -16,7 +16,7 @@ limitations under the License.
 
 package terminator
 
-type EventKind string
+type EventKind = string
 
 var EventKinds = struct {
 	AutoScalingTermination,
@@ -33,7 +33,3 @@ var EventKinds = struct {
 	StateChange:             EventKind("stateChange"),
 	Noop:                    EventKind("noop"),
 }
-
-func (e EventKind) String() string {
-	return string(e)
-}

src/pkg/terminator/reconciler.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/aws/aws-node-termination-handler/api/v1alpha1"
 	"github.com/aws/aws-node-termination-handler/pkg/logging"
+	"github.com/aws/aws-node-termination-handler/pkg/webhook"
 
 	"github.com/aws/aws-sdk-go/service/sqs"
 
@@ -47,6 +48,7 @@ type (
 	}
 
 	Event interface {
+		webhook.Event
 		zapcore.ObjectMarshaler
 
 		Done(context.Context) (tryAgain bool, err error)
@@ -83,13 +85,22 @@ type (
 		Parse(context.Context, *sqs.Message) Event
 	}
 
+	WebhookClient interface {
+		NewRequest() webhook.Request
+	}
+
+	WebhookClientBuilder interface {
+		NewWebhookClient(*v1alpha1.Terminator) (WebhookClient, error)
+	}
+
 	Reconciler struct {
+		CordonDrainerBuilder
+		Getter
 		NodeGetterBuilder
 		NodeNameGetter
 		SQSClientBuilder
 		SQSMessageParser
-		CordonDrainerBuilder
-		Getter
+		WebhookClientBuilder
 
 		Name            string
 		RequeueInterval time.Duration
@@ -109,6 +120,11 @@ func (r Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (recon
 
 	nodeGetter := r.NewNodeGetter(terminator)
 
+	webhookClient, err := r.NewWebhookClient(terminator)
+	if err != nil {
+		logging.FromContext(ctx).With("error", err).Warn("failed to initialize webhook client")
+	}
+
 	cordondrainer, err := r.NewCordonDrainer(terminator)
 	if err != nil {
 		return reconcile.Result{}, err
@@ -125,7 +141,7 @@ func (r Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (recon
 	}
 
 	for _, msg := range sqsMessages {
-		e := r.handleMessage(ctx, msg, terminator, nodeGetter, cordondrainer, sqsClient)
+		e := r.handleMessage(ctx, msg, terminator, nodeGetter, cordondrainer, sqsClient, webhookClient.NewRequest())
 		err = multierr.Append(err, e)
 	}
 
@@ -135,19 +151,22 @@ func (r Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (recon
 	return reconcile.Result{RequeueAfter: r.RequeueInterval}, nil
 }
 
-func (r Reconciler) handleMessage(ctx context.Context, msg *sqs.Message, terminator *v1alpha1.Terminator, nodeGetter NodeGetter, cordondrainer CordonDrainer, sqsClient SQSClient) (err error) {
+func (r Reconciler) handleMessage(ctx context.Context, msg *sqs.Message, terminator *v1alpha1.Terminator, nodeGetter NodeGetter, cordondrainer CordonDrainer, sqsClient SQSClient, webhookRequest webhook.Request) (err error) {
 	ctx = logging.WithLogger(ctx, logging.FromContext(ctx).With("sqsMessage", logging.NewMessageMarshaler(msg)))
 
 	evt := r.Parse(ctx, msg)
 	ctx = logging.WithLogger(ctx, logging.FromContext(ctx).With("event", evt))
 
+	webhookRequest.Event = evt
+
 	evtAction := actionForEvent(evt, terminator)
 	ctx = logging.WithLogger(ctx, logging.FromContext(ctx).With("action", evtAction))
 
 	allInstancesHandled := true
 	if evtAction != v1alpha1.Actions.NoAction {
 		for _, ec2InstanceID := range evt.EC2InstanceIDs() {
-			instanceHandled, e := r.handleInstance(ctx, ec2InstanceID, evtAction, nodeGetter, cordondrainer)
+			webhookRequest.InstanceID = ec2InstanceID
+			instanceHandled, e := r.handleInstance(ctx, ec2InstanceID, evtAction, nodeGetter, cordondrainer, webhookRequest)
 			err = multierr.Append(err, e)
 			allInstancesHandled = allInstancesHandled && instanceHandled
 		}
@@ -165,7 +184,7 @@ func (r Reconciler) handleMessage(ctx context.Context, msg *sqs.Message, termina
 	return multierr.Append(err, sqsClient.DeleteSQSMessage(ctx, msg))
 }
 
-func (r Reconciler) handleInstance(ctx context.Context, ec2InstanceID string, evtAction v1alpha1.Action, nodeGetter NodeGetter, cordondrainer CordonDrainer) (bool, error) {
+func (r Reconciler) handleInstance(ctx context.Context, ec2InstanceID string, evtAction v1alpha1.Action, nodeGetter NodeGetter, cordondrainer CordonDrainer, webhookRequest webhook.Request) (bool, error) {
 	ctx = logging.WithLogger(ctx, logging.FromContext(ctx).With("ec2InstanceID", ec2InstanceID))
 
 	nodeName, err := r.GetNodeName(ctx, ec2InstanceID)
@@ -185,6 +204,13 @@ func (r Reconciler) handleInstance(ctx context.Context, ec2InstanceID string, ev
 		return false, nil
 	}
 
+	webhookRequest.NodeName = nodeName
+	if err = webhookRequest.Send(ctx); err != nil {
+		logging.FromContext(ctx).
+			With("error", err).
+			Error("webhook notification failed")
+	}
+
 	if err = cordondrainer.Cordon(ctx, node); err != nil {
 		return true, err
 	}