Main scaffolding

Author: mortenlj

.gitignore

@@ -5,6 +5,7 @@
 .AppleDesktop
 .AppleDouble
 atlassian-ide-plugin.xml
+bin
 cmake-build-*/
 .com.apple.timemachine.donotpresent
 com_crashlytics_export_strings.xml
@@ -21,6 +22,7 @@ fabric.properties
 .fseventsd
 .fuse_hidden*
 gen
+go.sum
 Icon
 .idea
 .idea_modules/

Makefile

@@ -0,0 +1,5 @@
+aivenator:
+	go build -o bin/aivenator cmd/aivenator/*.go
+
+test:
+	go test ./... -count=1

cmd/aivenator/main.go

@@ -0,0 +1,169 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	aivenatormetrics "github.com/nais/aivenator/pkg/metrics"
+	"github.com/nais/liberator/pkg/apis/kafka.nais.io/v1"
+	log "github.com/sirupsen/logrus"
+	flag "github.com/spf13/pflag"
+	"github.com/spf13/viper"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+	// +kubebuilder:scaffold:imports
+)
+
+var scheme = runtime.NewScheme()
+
+type QuitChannel chan error
+
+const (
+	ExitOK = iota
+	ExitController
+	ExitConfig
+	ExitRuntime
+)
+
+// Configuration options
+const (
+	AivenToken                   = "aiven-token"
+	KubernetesWriteRetryInterval = "kubernetes-write-retry-interval"
+	LogFormat                    = "log-format"
+	MetricsAddress               = "metrics-address"
+	Projects                     = "projects"
+	SyncPeriod                   = "sync-period"
+)
+
+const (
+	LogFormatJSON = "json"
+	LogFormatText = "text"
+)
+
+func init() {
+
+	// Automatically read configuration options from environment variables.
+	// i.e. --aiven-token will be configurable using AIVENATOR_AIVEN_TOKEN.
+	viper.SetEnvPrefix("AIVENATOR")
+	viper.AutomaticEnv()
+	viper.SetEnvKeyReplacer(strings.NewReplacer("-", "_", ".", "_"))
+
+	flag.String(AivenToken, "", "Administrator credentials for Aiven")
+	flag.String(MetricsAddress, "127.0.0.1:8080", "The address the metric endpoint binds to.")
+	flag.String(LogFormat, "text", "Log format, either 'text' or 'json'")
+	flag.Duration(KubernetesWriteRetryInterval, time.Second*10, "Requeueing interval when Kubernetes writes fail")
+	flag.Duration(SyncPeriod, time.Hour*1, "How often to re-synchronize all AivenApplication resources including credential rotation")
+	flag.StringSlice(Projects, []string{"nav-integration-test"}, "List of projects allowed to operate on")
+
+	flag.Parse()
+
+	err := viper.BindPFlags(flag.CommandLine)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func formatter(logFormat string) (log.Formatter, error) {
+	switch logFormat {
+	case LogFormatJSON:
+		return &log.JSONFormatter{
+			TimestampFormat:   time.RFC3339Nano,
+			DisableHTMLEscape: true,
+		}, nil
+	case LogFormatText:
+		return &log.TextFormatter{
+			FullTimestamp:   true,
+			TimestampFormat: time.RFC3339Nano,
+		}, nil
+	}
+	return nil, fmt.Errorf("unsupported log format '%s'", logFormat)
+}
+
+func main() {
+	quit := make(QuitChannel, 1)
+	signals := make(chan os.Signal, 1)
+
+	logger := log.New()
+	logfmt, err := formatter(viper.GetString(LogFormat))
+	if err != nil {
+		logger.Error(err)
+		os.Exit(ExitConfig)
+	}
+
+	logger.SetFormatter(logfmt)
+
+	syncPeriod := viper.GetDuration(SyncPeriod)
+	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+		SyncPeriod:         &syncPeriod,
+		Scheme:             scheme,
+		MetricsBindAddress: viper.GetString(MetricsAddress),
+	})
+
+	if err != nil {
+		logger.Println(err)
+		os.Exit(ExitController)
+	}
+
+	logger.Info("Aivenator running")
+
+	go credentials(quit, logger, mgr)
+
+	go janitor(quit, logger, mgr)
+
+	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
+
+	go func() {
+		for {
+			select {
+			case err := <-quit:
+				logger.Errorf("terminating unexpectedly: %s", err)
+				os.Exit(ExitRuntime)
+			case sig := <-signals:
+				logger.Infof("exiting due to signal: %s", strings.ToUpper(sig.String()))
+				os.Exit(ExitOK)
+			}
+		}
+	}()
+
+	terminator := context.Background()
+	if err := mgr.Start(terminator); err != nil {
+		quit <- fmt.Errorf("manager stopped unexpectedly: %s", err)
+		return
+	}
+
+	quit <- fmt.Errorf("manager has stopped")
+}
+
+// TODO: Finds secrets that are no longer in use and cleans up associated service user before deleting secret
+func janitor(quit QuitChannel, logger *log.Logger, mgr manager.Manager) {
+
+}
+
+// TODO: Reconcile AivenApplications
+func credentials(quit QuitChannel, logger *log.Logger, mgr manager.Manager) {
+
+}
+
+func init() {
+	err := clientgoscheme.AddToScheme(scheme)
+	if err != nil {
+		panic(err)
+	}
+
+	err = kafka_nais_io_v1.AddToScheme(scheme)
+	if err != nil {
+		panic(err)
+	}
+
+	aivenatormetrics.Register(metrics.Registry)
+	// +kubebuilder:scaffold:scheme
+}

go.mod

@@ -0,0 +1,17 @@
+module github.com/nais/aivenator
+
+go 1.15
+
+require (
+	github.com/aiven/aiven-go-client v1.5.11
+	github.com/nais/liberator v0.0.0-20210421120235-5f3edcf81f86
+	github.com/prometheus/client_golang v1.7.1
+	github.com/sirupsen/logrus v1.8.1
+	github.com/spf13/pflag v1.0.5
+	github.com/spf13/viper v1.7.1
+	k8s.io/apimachinery v0.21.0
+	k8s.io/client-go v0.20.2
+	sigs.k8s.io/controller-runtime v0.8.3
+)
+
+replace github.com/nais/liberator => ../liberator

main.go

@@ -0,0 +1 @@
+package aivenator

pkg/metrics/metrics.go

@@ -0,0 +1,77 @@
+package metrics
+
+import (
+	"strconv"
+	"time"
+
+	"github.com/aiven/aiven-go-client"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	Namespace = "aivenator"
+
+	LabelAivenOperation = "operation"
+	LabelNamespace      = "namespace"
+	LabelPool           = "pool"
+	LabelResourceType   = "resource_type"
+	LabelStatus         = "status"
+	LabelSyncState      = "synchronization_state"
+)
+
+var (
+	TopicsProcessed = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name:      "aiven_applications_processed",
+		Namespace: Namespace,
+		Help:      "number of applications synchronized with aiven",
+	}, []string{LabelSyncState, LabelPool})
+
+	ServiceUsers = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:      "service_users",
+		Namespace: Namespace,
+		Help:      "number of service users",
+	}, []string{LabelPool})
+
+	AivenLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:      "aiven_latency",
+		Namespace: Namespace,
+		Help:      "latency in aiven api operations",
+		Buckets:   prometheus.LinearBuckets(0.05, 0.05, 100),
+	}, []string{LabelAivenOperation, LabelStatus, LabelPool})
+
+	KubernetesResourcesWritten = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name:      "kubernetes_resources_written",
+		Namespace: Namespace,
+		Help:      "number of kubernetes resources written to the cluster",
+	}, []string{LabelNamespace, LabelResourceType})
+)
+
+func ObserveAivenLatency(operation, pool string, fun func() error) error {
+	timer := time.Now()
+	err := fun()
+	used := time.Now().Sub(timer)
+	status := 200
+	if err != nil {
+		aivenErr, ok := err.(aiven.Error)
+		if ok {
+			status = aivenErr.Status
+		} else {
+			status = 0
+		}
+	}
+	AivenLatency.With(prometheus.Labels{
+		LabelAivenOperation: operation,
+		LabelPool:           pool,
+		LabelStatus:         strconv.Itoa(status),
+	}).Observe(used.Seconds())
+	return err
+}
+
+func Register(registry prometheus.Registerer) {
+	registry.MustRegister(
+		AivenLatency,
+		KubernetesResourcesWritten,
+		ServiceUsers,
+		TopicsProcessed,
+	)
+}