Adding SNS Topic Creation

**Why:**

* allows you to create SNS topics via `kubectl`

**This change addresses the need by:**

* closes #33

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

Author: christopherhein

examples/snstopic.yaml

@@ -0,0 +1,4 @@
+apiVersion: operator.aws/v1alpha1
+kind: SNSTopic
+metadata:
+  name: sns-topic-10

models/snstopic.yaml

@@ -0,0 +1,31 @@
+apiVersion: operator.aws/v1alpha1
+kind: ModelDefinition
+metadata:
+  name: SNSResource
+spec:
+  kind: SNSTopic
+  type: Spec # can be Spec or Data
+  queue: true
+  useCloudFormation: true
+  resource:
+    name: snstopic
+    plural: snstopics
+    shortNames:
+    - name: sns
+    - name: topic
+    - name: topics
+  body:
+    schema:
+      type: object
+      properties:
+
+  output:
+    schema:
+      type: object
+      properties:
+      - key: topicName
+        type: string
+        description: |
+          TopicName is the output topicname incase it changed
+        structKey: TopicName
+        templateKey: TopicName

pkg/helpers/helpers.go

@@ -5,7 +5,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/cloudformation"
 	awsclient "github.com/christopherhein/aws-operator/pkg/client/clientset/versioned/typed/operator.aws/v1alpha1"
 	"github.com/christopherhein/aws-operator/pkg/config"
-	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"reflect"
 	"strconv"
@@ -69,24 +68,6 @@ func Templatize(tempStr string, data interface{}) (resp string, err error) {
 	return tpl.String(), err
 }
 
-// ServiceType will return the service object
-func ServiceType(str string) apiv1.ServiceType {
-	var sType apiv1.ServiceType
-	switch str {
-	case "ClusterIP":
-		sType = apiv1.ServiceTypeClusterIP
-	case "NodePort":
-		sType = apiv1.ServiceTypeNodePort
-	case "LoadBalancer":
-		sType = apiv1.ServiceTypeLoadBalancer
-	case "ExternalName":
-		sType = apiv1.ServiceTypeExternalName
-	default:
-		sType = apiv1.ServiceTypeClusterIP
-	}
-	return sType
-}
-
 // GetCloudFormationTemplate will return the url to the CFT from the CFT resource
 func GetCloudFormationTemplate(config *config.Config, rType string, name string, namespace string) string {
 	logger := config.Logger

pkg/helpers/service.go

@@ -0,0 +1,61 @@
+package helpers
+
+import (
+	"github.com/christopherhein/aws-operator/pkg/config"
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type Data struct {
+	Obj    interface{}
+	Config *config.Config
+}
+
+// CreateExternalNameService will create a Kubernetes Servic Using ExternalName types
+func CreateExternalNameService(config *config.Config, resource interface{}, svcName string, svcNamespace string, externalNameTemplate string, svcPort int32) string {
+	logger := config.Logger
+
+	externalName, err := Templatize(externalNameTemplate, Data{Obj: resource, Config: config})
+	if err != nil {
+		logger.WithError(err).Error("error parsing external name template")
+		return ""
+	}
+
+	service := &apiv1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: svcName,
+		},
+		Spec: apiv1.ServiceSpec{
+			Type:         apiv1.ServiceTypeExternalName,
+			ExternalName: externalName,
+			Ports: []apiv1.ServicePort{
+				apiv1.ServicePort{
+					Port: svcPort,
+				},
+			},
+		},
+	}
+
+	newService, err := config.Context.Clientset.CoreV1().Services(svcNamespace).Create(service)
+	if err != nil {
+		logger.WithError(err).Error("error creating service")
+	}
+	return newService.Name
+}
+
+// func serviceType(str string) apiv1.ServiceType {
+// 	var sType apiv1.ServiceType
+// 	switch str {
+// 	case "ClusterIP":
+// 		sType = apiv1.ServiceTypeClusterIP
+// 	case "NodePort":
+// 		sType = apiv1.ServiceTypeNodePort
+// 	case "LoadBalancer":
+// 		sType = apiv1.ServiceTypeLoadBalancer
+// 	case "ExternalName":
+// 		sType = apiv1.ServiceTypeExternalName
+// 	default:
+// 		sType = apiv1.ServiceTypeClusterIP
+// 	}
+// 	return sType
+// }

pkg/queue/queue.go

@@ -224,6 +224,7 @@ func process(q *Queue, svc *sqs.SQS, h Handler, stopCh <-chan struct{}) error {
 				logger.WithError(err).Error("error parsing message body")
 			}
 			mb.ParseMessage()
+			logger.Debugf("%+v", mb)
 
 			err = h.HandleMessage(config, mb)
 			if err != nil {

pkg/server/server.go

@@ -12,6 +12,7 @@ import (
 	"github.com/christopherhein/aws-operator/pkg/operator/cloudformationtemplate"
 	"github.com/christopherhein/aws-operator/pkg/operator/dynamodb"
 	"github.com/christopherhein/aws-operator/pkg/operator/s3bucket"
+	"github.com/christopherhein/aws-operator/pkg/operator/snstopic"
 	"github.com/christopherhein/aws-operator/pkg/operator/sqsqueue"
 	opkit "github.com/christopherhein/operator-kit"
 	"k8s.io/api/core/v1"
@@ -48,6 +49,7 @@ func (c *Server) Run(stopChan chan struct{}) {
 		s3bucket.Resource,
 		dynamodb.Resource,
 		sqsqueue.Resource,
+		snstopic.Resource,
 	}
 	err = opkit.CreateCustomResources(*context, resources)
 	if err != nil {
@@ -91,6 +93,9 @@ func (c *Server) Run(stopChan chan struct{}) {
 
 	sqscontroller := sqsqueue.NewController(config, context, awsClientset)
 	sqscontroller.StartWatch(v1.NamespaceAll, stopChan)
+
+	snscontroller := snstopic.NewController(config, context, awsClientset)
+	snscontroller.StartWatch(v1.NamespaceAll, stopChan)
 }
 
 func getClientConfig(kubeconfig string) (*rest.Config, error) {