Merge pull request #5 from rusenask/develop

Develop

Dockerfile

@@ -1,6 +1,12 @@
+FROM golang:1.8.1-alpine
+COPY . /go/src/github.com/rusenask/keel
+WORKDIR /go/src/github.com/rusenask/keel
+RUN apk add --no-cache git && go get
+RUN CGO_ENABLED=0 GOOS=linux go build -a -tags netgo  -ldflags  -'w' -o keel .
+
 FROM alpine:latest
 RUN apk --no-cache add ca-certificates
-COPY       keel /bin/keel
+COPY --from=0 /go/src/github.com/rusenask/keel/keel /bin/keel
 ENTRYPOINT ["/bin/keel"]
 
 EXPOSE 9300

glide.yaml

@@ -1,11 +1,52 @@
 package: github.com/rusenask/keel
 import:
+- package: cloud.google.com/go
+  version: b4e9a381a01e953e880e6d2cf7fd02d412977cae
+  subpackages:
+  - pubsub
+  - iam
+  - internal
+- package: google.golang.org/api
+  version: 324744a33f1f37e63dd1695cfb3ec9a3e4a1cb05
+  subpackages:
+  - iterator
+  - option
+  - support  
+  - transport
+- package: google.golang.org/grpc
+  version: a94a7ac054dc76e5ab6cf170a9d82faba8aaf33f
+  subpackages:
+  - status  
+- package: github.com/Sirupsen/logrus  
+  subpackages:
+  - formatters
+  - formatters/logstash
+- package: github.com/coreos/go-semver
+  subpackages:
+  - semver
+- package: github.com/gorilla/mux
+- package: github.com/urfave/negroni
+- package: golang.org/x/net
+  subpackages:
+  - context
+  - oauth2
+- package: github.com/gogo/protobuf
+  version: c0656edd0d9eab7c66d1eb0c568f9039345796f7
+  subpackages:
+  - sortkeys
+  - protoc-gen-gogo/descriptor
+  - gogoproto
+  - plugin/testgen
+  - protoc-gen-gogo/generator  
+- package: k8s.io/apimachinery
+  version: 75b8dd260ef0469d96d578705a87cffd0e09dab8
+  subpackages:
+  - pkg/apis/meta/v1
 - package: k8s.io/client-go
   version: v3.0.0-beta.0
   subpackages:
   - kubernetes
   - pkg/api/v1
   - pkg/apis/extensions/v1beta1
   - rest
-- package: k8s.io/apimachinery
-  version: 75b8dd260ef0469d96d578705a87cffd0e09dab8
+  - tools/clientcmd

main.go

@@ -9,8 +9,8 @@ import (
 
 	"github.com/rusenask/keel/provider"
 	"github.com/rusenask/keel/provider/kubernetes"
+	"github.com/rusenask/keel/trigger/http"
 	"github.com/rusenask/keel/trigger/pubsub"
-	"github.com/rusenask/keel/trigger/webhook"
 	"github.com/rusenask/keel/types"
 
 	log "github.com/Sirupsen/logrus"
@@ -51,28 +51,81 @@ func main() {
 		}).Fatal("main: failed to create kubernetes implementer")
 	}
 
-	k8sProvider, err := kubernetes.NewProvider(implementer)
+	// setting up providers
+	providers, teardownProviders := setupProviders(implementer)
+
+	// setting up triggers
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	teardownTriggers := setupTriggers(ctx, implementer, providers)
+
+	signalChan := make(chan os.Signal, 1)
+	cleanupDone := make(chan bool)
+	signal.Notify(signalChan, os.Interrupt)
+	go func() {
+		for _ = range signalChan {
+			log.Info("received an interrupt, closing connection...")
+
+			go func() {
+				select {
+				case <-time.After(10 * time.Second):
+					log.Info("connection shutdown took too long, exiting... ")
+					close(cleanupDone)
+					return
+				case <-cleanupDone:
+					return
+				}
+			}()
+
+			teardownProviders()
+			teardownTriggers()
+
+			cleanupDone <- true
+		}
+	}()
+
+	<-cleanupDone
+
+}
+
+// setupProviders - setting up available providers. New providers should be initialised here and added to
+// provider map
+func setupProviders(k8sImplementer kubernetes.Implementer) (providers map[string]provider.Provider, teardown func()) {
+	k8sProvider, err := kubernetes.NewProvider(k8sImplementer)
 	if err != nil {
 		log.WithFields(log.Fields{
 			"error": err,
-		}).Fatal("main: failed to create kubernetes provider")
+		}).Fatal("main.setupProviders: failed to create kubernetes provider")
 	}
 	go k8sProvider.Start()
 
-	providers := make(map[string]provider.Provider)
+	providers = make(map[string]provider.Provider)
 	providers[k8sProvider.GetName()] = k8sProvider
 
-	whs := webhook.NewTriggerServer(&webhook.Opts{
+	teardown = func() {
+		k8sProvider.Stop()
+	}
+
+	return providers, teardown
+}
+
+// setupTriggers - setting up triggers. New triggers should be added to this function. Each trigger
+// should go through all providers (or not if there is a reason) and submit events)
+func setupTriggers(ctx context.Context, k8sImplementer kubernetes.Implementer, providers map[string]provider.Provider) (teardown func()) {
+
+	// setting up generic http webhook server
+	whs := http.NewTriggerServer(&http.Opts{
 		Port:      types.KeelDefaultPort,
 		Providers: providers,
 	})
 
 	go whs.Start()
 
+	// checking whether pubsub (GCR) trigger is enabled
 	if os.Getenv(EnvTriggerPubSub) != "" {
 		projectID := os.Getenv(EnvProjectID)
 		if projectID == "" {
-			log.Fatalf("main: project ID env variable not set")
+			log.Fatalf("main.setupTriggers: project ID env variable not set")
 			return
 		}
 
@@ -83,40 +136,17 @@ func main() {
 		if err != nil {
 			log.WithFields(log.Fields{
 				"error": err,
-			}).Fatal("main: failed to create gcloud pubsub subscriber")
+			}).Fatal("main.setupTriggers: failed to create gcloud pubsub subscriber")
 			return
 		}
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
 
-		subManager := pubsub.NewDefaultManager(projectID, implementer, ps)
+		subManager := pubsub.NewDefaultManager(projectID, k8sImplementer, ps)
 		go subManager.Start(ctx)
 	}
 
-	signalChan := make(chan os.Signal, 1)
-	cleanupDone := make(chan bool)
-	signal.Notify(signalChan, os.Interrupt)
-	go func() {
-		for _ = range signalChan {
-			log.Info("received an interrupt, closing connection...")
-
-			go func() {
-				select {
-				case <-time.After(10 * time.Second):
-					log.Info("connection shutdown took too long, exiting... ")
-					close(cleanupDone)
-					return
-				case <-cleanupDone:
-					return
-				}
-			}()
-
-			k8sProvider.Stop()
-			// whs.Stop()
-			cleanupDone <- true
-		}
-	}()
-
-	<-cleanupDone
+	teardown = func() {
+		whs.Stop()
+	}
 
+	return teardown
 }

provider/kubernetes/implementer.go

@@ -21,24 +21,23 @@ type Implementer interface {
 	Update(deployment *v1beta1.Deployment) error
 }
 
+// KubernetesImplementer - default kubernetes client implementer, uses
+// https://github.com/kubernetes/client-go v3.0.0-beta.0
 type KubernetesImplementer struct {
 	cfg    *rest.Config
 	client *kubernetes.Clientset
 }
 
+// Opts - implementer options, usually for k8s deployments
+// it's best to use InCluster option
 type Opts struct {
 	// if set - kube config options will be ignored
-	InCluster bool
-
+	InCluster  bool
 	ConfigPath string
-
-	// Master host
-	Master   string
-	KeyFile  string
-	CAFile   string
-	CertFile string
+	Master     string
 }
 
+// NewKubernetesImplementer - create new k8s implementer
 func NewKubernetesImplementer(opts *Opts) (*KubernetesImplementer, error) {
 	cfg := &rest.Config{}
 
@@ -76,22 +75,26 @@ func NewKubernetesImplementer(opts *Opts) (*KubernetesImplementer, error) {
 	return &KubernetesImplementer{client: client, cfg: cfg}, nil
 }
 
+// Namespaces - get all namespaces
 func (i *KubernetesImplementer) Namespaces() (*v1.NamespaceList, error) {
 	namespaces := i.client.Namespaces()
 	return namespaces.List(meta_v1.ListOptions{})
 }
 
+// Deployment - get specific deployment for namespace/name
 func (i *KubernetesImplementer) Deployment(namespace, name string) (*v1beta1.Deployment, error) {
 	dep := i.client.Extensions().Deployments(namespace)
 	return dep.Get(name, meta_v1.GetOptions{})
 }
 
+// Deployments - get all deployments for namespace
 func (i *KubernetesImplementer) Deployments(namespace string) (*v1beta1.DeploymentList, error) {
 	dep := i.client.Extensions().Deployments(namespace)
 	l, err := dep.List(meta_v1.ListOptions{})
 	return l, err
 }
 
+// Update - update deployment
 func (i *KubernetesImplementer) Update(deployment *v1beta1.Deployment) error {
 	_, err := i.client.Extensions().Deployments(deployment.Namespace).Update(deployment)
 	return err

provider/kubernetes/kubernetes.go

@@ -13,15 +13,13 @@ import (
 	log "github.com/Sirupsen/logrus"
 )
 
+// ProviderName - provider name
 const ProviderName = "kubernetes"
 
 var versionreg = regexp.MustCompile(`:[^:]*$`)
 
 // Provider - kubernetes provider for auto update
 type Provider struct {
-	// cfg    *rest.Config
-	// client *kubernetes.Clientset
-
 	implementer Implementer
 
 	events chan *types.Event
@@ -31,8 +29,6 @@ type Provider struct {
 // NewProvider - create new kubernetes based provider
 func NewProvider(implementer Implementer) (*Provider, error) {
 	return &Provider{
-		// cfg:    cfg,
-		// client: client,
 		implementer: implementer,
 		events:      make(chan *types.Event, 100),
 		stop:        make(chan struct{}),
@@ -96,15 +92,12 @@ func (p *Provider) processEvent(event *types.Event) (updated []*v1beta1.Deployme
 		return
 	}
 
-	updated, err = p.updateDeployments(impacted)
-
-	return
+	return p.updateDeployments(impacted)
 }
 
-func (p *Provider) updateDeployments(deployments []*v1beta1.Deployment) (updated []*v1beta1.Deployment, err error) {
+func (p *Provider) updateDeployments(deployments []v1beta1.Deployment) (updated []*v1beta1.Deployment, err error) {
 	for _, deployment := range deployments {
-		// _, err := p.client.Extensions().Deployments(deployment.Namespace).Update(deployment)
-		err := p.implementer.Update(deployment)
+		err := p.implementer.Update(&deployment)
 		if err != nil {
 			log.WithFields(log.Fields{
 				"error":      err,
@@ -113,7 +106,11 @@ func (p *Provider) updateDeployments(deployments []*v1beta1.Deployment) (updated
 			}).Error("provider.kubernetes: got error while update deployment")
 			continue
 		}
-		updated = append(updated, deployment)
+		log.WithFields(log.Fields{
+			"name":      deployment.Name,
+			"namespace": deployment.Namespace,
+		}).Info("provider.kubernetes: deployment updated")
+		updated = append(updated, &deployment)
 	}
 
 	return
@@ -127,7 +124,7 @@ func (p *Provider) getDeployment(namespace, name string) (*v1beta1.Deployment, e
 }
 
 // gets impacted deployments by changed repository
-func (p *Provider) impactedDeployments(repo *types.Repository) ([]*v1beta1.Deployment, error) {
+func (p *Provider) impactedDeployments(repo *types.Repository) ([]v1beta1.Deployment, error) {
 	newVersion, err := version.GetVersion(repo.Tag)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse version from repository tag, error: %s", err)
@@ -141,7 +138,7 @@ func (p *Provider) impactedDeployments(repo *types.Repository) ([]*v1beta1.Deplo
 		return nil, err
 	}
 
-	impacted := []*v1beta1.Deployment{}
+	impacted := []v1beta1.Deployment{}
 
 	for _, deploymentList := range deploymentLists {
 		for _, deployment := range deploymentList.Items {
@@ -223,7 +220,7 @@ func (p *Provider) impactedDeployments(repo *types.Repository) ([]*v1beta1.Deplo
 					// updating image
 					c.Image = fmt.Sprintf("%s:%s", containerImageName, newVersion.String())
 					deployment.Spec.Template.Spec.Containers[idx] = c
-					impacted = append(impacted, &deployment)
+					impacted = append(impacted, deployment)
 					log.WithFields(log.Fields{
 						"parsed_image":     containerImageName,
 						"raw_image_name":   c.Image,
@@ -240,14 +237,11 @@ func (p *Provider) impactedDeployments(repo *types.Repository) ([]*v1beta1.Deplo
 }
 
 func (p *Provider) namespaces() (*v1.NamespaceList, error) {
-	// namespaces := p.client.Namespaces()
-	// return namespaces.List(meta_v1.ListOptions{})
 	return p.implementer.Namespaces()
 }
 
 // deployments - gets all deployments
 func (p *Provider) deployments() ([]*v1beta1.DeploymentList, error) {
-	// namespaces := p.client.Namespaces()
 	deployments := []*v1beta1.DeploymentList{}
 
 	n, err := p.namespaces()
@@ -256,8 +250,6 @@ func (p *Provider) deployments() ([]*v1beta1.DeploymentList, error) {
 	}
 
 	for _, n := range n.Items {
-		// dep := p.client.Extensions().Deployments(n.GetName())
-		// l, err := dep.List(meta_v1.ListOptions{})
 		l, err := p.implementer.Deployments(n.GetName())
 		if err != nil {
 			log.WithFields(log.Fields{