Adding Kubernetes Recorder to log to the standard events

* Adds event recorder which will log status updates to the event stream
* Will continue between rollbacks so you can see it failing

closes #85

Signed-off-by: Christopher Hein <[email protected]>

Author: christopherhein

Makefile

@@ -28,7 +28,7 @@ aws-codegen:
 
 .PHONY: k8s-codegen
 k8s-codegen:
-	./codegen.sh
+	./hack/update-codegen.sh
 
 .PHONY: codegen
 codegen: aws-codegen k8s-codegen

code-generation/pkg/codegen/assets/controller.go.templ

@@ -14,7 +14,10 @@ import (
 
 	"github.com/awslabs/aws-service-operator/pkg/config"
 	{{- if .Spec.Queue}}
-  "github.com/awslabs/aws-service-operator/pkg/queue"
+    "github.com/awslabs/aws-service-operator/pkg/queue"
+    corev1 "k8s.io/api/core/v1"
+    "github.com/iancoleman/strcase"
+    "strings"
 	{{- end}}
 	opkit "github.com/christopherhein/operator-kit"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
@@ -50,7 +53,7 @@ type Controller struct {
 	config       *config.Config
 	context      *opkit.Context
 	awsclientset awsclient.ServiceoperatorV1alpha1Interface
-  topicARN     string
+	topicARN     string
 }
 
 // NewController create controller for watching object store custom resources created
@@ -106,27 +109,35 @@ func QueueUpdater(config *config.Config, msg *queue.MessageBody) error {
 	}
 
 	if name != "" && namespace != "" {
+		annotations := map[string]string{
+			"StackID": msg.ParsedMessage["StackId"],
+			"StackName": msg.ParsedMessage["StackName"],
+			"ResourceType": msg.ParsedMessage["ResourceType"],
+		}
 		if msg.ParsedMessage["ResourceStatus"] == "ROLLBACK_COMPLETE" {
-			err := deleteStack(config, name, namespace, msg.ParsedMessage["StackId"])
+			obj, err := deleteStack(config, name, namespace, msg.ParsedMessage["StackId"])
 			if err != nil {
 				return err
 			}
+			config.Recorder.AnnotatedEventf(obj, annotations, corev1.EventTypeWarning, strcase.ToCamel(strings.ToLower(msg.ParsedMessage["ResourceStatus"])), msg.ParsedMessage["ResourceStatusReason"])
 		} else if msg.ParsedMessage["ResourceStatus"] == "DELETE_COMPLETE" {
-			err := updateStatus(config, name, namespace, msg.ParsedMessage["StackId"], msg.ParsedMessage["ResourceStatus"], msg.ParsedMessage["ResourceStatusReason"])
+			obj, err := updateStatus(config, name, namespace, msg.ParsedMessage["StackId"], msg.ParsedMessage["ResourceStatus"], msg.ParsedMessage["ResourceStatusReason"])
 			if err != nil {
 				return err
 			}
-
+			config.Recorder.AnnotatedEventf(obj, annotations, corev1.EventTypeWarning, strcase.ToCamel(strings.ToLower(msg.ParsedMessage["ResourceStatus"])), msg.ParsedMessage["ResourceStatusReason"])
 			err = incrementRollbackCount(config, name, namespace)
 			if err != nil {
 				return err
 			}
 		} else {
-			err := updateStatus(config, name, namespace, msg.ParsedMessage["StackId"], msg.ParsedMessage["ResourceStatus"], msg.ParsedMessage["ResourceStatusReason"])
+			obj, err := updateStatus(config, name, namespace, msg.ParsedMessage["StackId"], msg.ParsedMessage["ResourceStatus"], msg.ParsedMessage["ResourceStatusReason"])
 			if err != nil {
 				return err
 			}
+			config.Recorder.AnnotatedEventf(obj, annotations, corev1.EventTypeNormal, strcase.ToCamel(strings.ToLower(msg.ParsedMessage["ResourceStatus"])), msg.ParsedMessage["ResourceStatusReason"])
 		}
+		
 	}
 
 	return nil
@@ -147,7 +158,7 @@ func (c *Controller) onAdd(obj interface{}) {
     c.config.Logger.Infof("added {{.Spec.Resource.Name}} '%s' with stackID '%s'", s.Name, string(*output.StackId))
     c.config.Logger.Infof("view at https://console.aws.amazon.com/cloudformation/home?#/stack/detail?stackId=%s", string(*output.StackId))
 
-		err = updateStatus(c.config, s.Name, s.Namespace, string(*output.StackId), "CREATE_IN_PROGRESS", "")
+		_, err = updateStatus(c.config, s.Name, s.Namespace, string(*output.StackId), "CREATE_IN_PROGRESS", "")
 		if err != nil {
 			c.config.Logger.WithError(err).Error("error updating status")
 		}
@@ -176,7 +187,7 @@ func (c *Controller) onUpdate(oldObj, newObj interface{}) {
     c.config.Logger.Infof("updated {{.Spec.Resource.Name}} '%s' with params '%s'", no.Name, string(*output.StackId))
     c.config.Logger.Infof("view at https://console.aws.amazon.com/cloudformation/home?#/stack/detail?stackId=%s", string(*output.StackId))
 
-		err = updateStatus(c.config, oo.Name, oo.Namespace, string(*output.StackId), "UPDATE_IN_PROGRESS", "")
+		_, err = updateStatus(c.config, oo.Name, oo.Namespace, string(*output.StackId), "UPDATE_IN_PROGRESS", "")
 		if err != nil {
 			c.config.Logger.WithError(err).Error("error updating status")
 		}
@@ -225,13 +236,13 @@ func incrementRollbackCount(config *config.Config, name string, namespace string
 	return nil
 }
 
-func updateStatus(config *config.Config, name string, namespace string, stackID string, status string, reason string) error {
+func updateStatus(config *config.Config, name string, namespace string, stackID string, status string, reason string) (*awsV1alpha1.{{.Spec.Kind}}, error) {
 	logger := config.Logger
 	clientSet, _ := awsclient.NewForConfig(config.RESTConfig)
 	resource, err := clientSet.{{.Spec.PluralName}}(namespace).Get(name, metav1.GetOptions{})
 	if err != nil {
 		logger.WithError(err).Error("error getting {{.Spec.Resource.Plural}}")
-		return err
+		return nil, err
 	}
 
 	resourceCopy := resource.DeepCopy()
@@ -259,7 +270,7 @@ func updateStatus(config *config.Config, name string, namespace string, stackID
 	_, err = clientSet.{{.Spec.PluralName}}(namespace).Update(resourceCopy)
 	if err != nil {
 		logger.WithError(err).Error("error updating resource")
-		return err
+		return nil, err
 	}
 
 	if helpers.IsStackComplete(status, false) {
@@ -268,26 +279,26 @@ func updateStatus(config *config.Config, name string, namespace string, stackID
 			logger.WithError(err).Info("error syncing resources")
 		}
 	}
-	return nil
+	return resourceCopy, nil
 }
 
-func deleteStack(config *config.Config, name string, namespace string, stackID string) error {
+func deleteStack(config *config.Config, name string, namespace string, stackID string) (*awsV1alpha1.{{.Spec.Kind}}, error) {
 	logger := config.Logger
 	clientSet, _ := awsclient.NewForConfig(config.RESTConfig)
 	resource, err := clientSet.{{.Spec.PluralName}}(namespace).Get(name, metav1.GetOptions{})
 	if err != nil {
 		logger.WithError(err).Error("error getting {{.Spec.Resource.Plural}}")
-		return err
+		return nil, err
 	}
 
 	cft := New(config, resource, "")
 	err = cft.DeleteStack()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	err = cft.WaitUntilStackDeleted()
-	return err
+	return resource, err
 }
 
 func syncAdditionalResources(config *config.Config, s *awsV1alpha1.{{.Spec.Kind}}) (err error) {

code-generation/pkg/codegen/templates.go

@@ -0,0 +1,13 @@
+/*
+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.
+*/

codegen.sh

@@ -0,0 +1,17 @@
+set -o errexit
+set -o nounset
+set -o pipefail
+
+SCRIPT_ROOT=$(dirname ${BASH_SOURCE})/..
+CODEGEN_PKG=${CODEGEN_PKG:-$(cd ${SCRIPT_ROOT}; ls -d -1 ./vendor/k8s.io/code-generator 2>/dev/null || echo ../code-generator)}
+
+# generate the code with:
+# --output-base    because this script should also be able to run inside the vendor dir of
+#                  k8s.io/kubernetes. The output-base is needed for the generators to output into the vendor dir
+#                  instead of the $GOPATH directly. For normal projects this can be dropped.
+${CODEGEN_PKG}/generate-groups.sh all \
+  github.com/awslabs/aws-service-operator/pkg/client \
+  github.com/awslabs/aws-service-operator/pkg/apis \
+  "service-operator.aws:v1alpha1" \
+  --output-base "$(dirname ${BASH_SOURCE})/../../.." \
+  --go-header-file ${SCRIPT_ROOT}/hack/boilerplate.go.txt

hack/boilerplate.go.txt

@@ -6,6 +6,7 @@ import (
 	opkit "github.com/christopherhein/operator-kit"
 	"github.com/sirupsen/logrus"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
 )
 
 // Config defines the configuration for the operator
@@ -22,6 +23,7 @@ type Config struct {
 	ClusterName   string
 	Bucket        string
 	AccountID     string
+	Recorder	  record.EventRecorder
 }
 
 // LoggingConfig defines the attributes for the logger

hack/update-codegen.sh

@@ -230,7 +230,7 @@ func process(q *Queue, svc *sqs.SQS, h Handler, stopCh <-chan struct{}) error {
 			if err != nil {
 				logger.WithError(err).Error("error processing message")
 			}
-			logger.Infof("stackID %v updated status to %v", mb.ParsedMessage["StackId"], mb.ParsedMessage["ResourceStatus"])
+			logger.Debugf("stackID %v updated status to %v", mb.ParsedMessage["StackId"], mb.ParsedMessage["ResourceStatus"])
 			_, err = svc.DeleteMessage(&sqs.DeleteMessageInput{
 				QueueUrl:      aws.String(q.queueURL),
 				ReceiptHandle: message.ReceiptHandle,

pkg/config/types.go

@@ -17,13 +17,19 @@ import (
 	"github.com/awslabs/aws-service-operator/pkg/operator/snstopic"
 	"github.com/awslabs/aws-service-operator/pkg/operator/sqsqueue"
 	opkit "github.com/christopherhein/operator-kit"
-	"k8s.io/api/core/v1"
+	corev1 "k8s.io/api/core/v1"
 	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/tools/record"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	awsscheme "github.com/awslabs/aws-service-operator/pkg/client/clientset/versioned/scheme"
 )
 
+const controllerName = "aws-service-operator"
+
 // New creates a new server from a config
 func New(config *config.Config) *Server {
 	return &Server{
@@ -36,7 +42,7 @@ func (c *Server) Run(stopChan chan struct{}) {
 	config := c.Config
 	logger := config.Logger
 	logger.Info("Getting kubernetes context")
-	context, restConfig, awsClientset, err := createContext(config.Kubeconfig)
+	context, restConfig, kubeclientset, awsClientset, err := createContext(config.Kubeconfig)
 	if err != nil {
 		logger.Fatalf("failed to create context. %+v\n", err)
 	}
@@ -84,28 +90,35 @@ func (c *Server) Run(stopChan chan struct{}) {
 	}
 	config.AWSSession = sess
 
+	awsscheme.AddToScheme(scheme.Scheme)
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(logger.Infof)
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events("")})
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName})
+	config.Recorder = recorder
+
 	// start watching the aws operator resources
 	logger.Info("Watching the resources")
 	cftcontroller := cloudformationtemplate.NewController(config, context, awsClientset)
-	cftcontroller.StartWatch(v1.NamespaceAll, stopChan)
+	cftcontroller.StartWatch(corev1.NamespaceAll, stopChan)
 
 	s3controller := s3bucket.NewController(config, context, awsClientset)
-	s3controller.StartWatch(v1.NamespaceAll, stopChan)
+	s3controller.StartWatch(corev1.NamespaceAll, stopChan)
 
 	ddbcontroller := dynamodb.NewController(config, context, awsClientset)
-	ddbcontroller.StartWatch(v1.NamespaceAll, stopChan)
+	ddbcontroller.StartWatch(corev1.NamespaceAll, stopChan)
 
 	sqscontroller := sqsqueue.NewController(config, context, awsClientset)
-	sqscontroller.StartWatch(v1.NamespaceAll, stopChan)
+	sqscontroller.StartWatch(corev1.NamespaceAll, stopChan)
 
 	ecrcontroller := ecrrepository.NewController(config, context, awsClientset)
-	ecrcontroller.StartWatch(v1.NamespaceAll, stopChan)
+	ecrcontroller.StartWatch(corev1.NamespaceAll, stopChan)
 
 	snscontroller := snstopic.NewController(config, context, awsClientset)
-	snscontroller.StartWatch(v1.NamespaceAll, stopChan)
+	snscontroller.StartWatch(corev1.NamespaceAll, stopChan)
 
 	snssubcontroller := snssubscription.NewController(config, context, awsClientset)
-	snssubcontroller.StartWatch(v1.NamespaceAll, stopChan)
+	snssubcontroller.StartWatch(corev1.NamespaceAll, stopChan)
 }
 
 func getClientConfig(kubeconfig string) (*rest.Config, error) {
@@ -115,25 +128,25 @@ func getClientConfig(kubeconfig string) (*rest.Config, error) {
 	return rest.InClusterConfig()
 }
 
-func createContext(kubeconfig string) (*opkit.Context, *rest.Config, awsclient.ServiceoperatorV1alpha1Interface, error) {
+func createContext(kubeconfig string) (*opkit.Context, *rest.Config, kubernetes.Interface, awsclient.ServiceoperatorV1alpha1Interface, error) {
 	config, err := getClientConfig(kubeconfig)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to get k8s config. %+v", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to get k8s config. %+v", err)
 	}
 
 	clientset, err := kubernetes.NewForConfig(config)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to get k8s client. %+v", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to get k8s client. %+v", err)
 	}
 
 	apiExtClientset, err := apiextensionsclient.NewForConfig(config)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to create k8s API extension clientset. %+v", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create k8s API extension clientset. %+v", err)
 	}
 
 	awsclientset, err := awsclient.NewForConfig(config)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to create object store clientset. %+v", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create object store clientset. %+v", err)
 	}
 
 	context := &opkit.Context{
@@ -142,5 +155,5 @@ func createContext(kubeconfig string) (*opkit.Context, *rest.Config, awsclient.S
 		Interval:              500 * time.Millisecond,
 		Timeout:               60 * time.Second,
 	}
-	return context, config, awsclientset, nil
+	return context, config, clientset, awsclientset, nil
 }