MVP

Dockerfile

@@ -1,38 +1,15 @@
-# Start from the latest golang base image
-FROM golang:latest as builder
-
-# Add Maintainer Info
-LABEL maintainer="Borisav Zivanovic <borisavzivanovic@gmail.com>"
-
-# Set the Current Working Directory inside the container
+FROM golang:alpine as build_container
 WORKDIR /app
-
-# Copy go mod and sum files
-COPY go.mod go.sum ./
-
-
-# Download all dependencies. Dependencies will be cached if the go.mod and go.sum files are not changed
+COPY ./go.mod ./go.sum ./
 RUN go mod download
+COPY ./ .
+RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o configServer .
 
-# Copy everything from the current directory to the Working Directory inside the container
-COPY . .
-
-# Build the Go app
-RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o main .
-
-
-
-######## Start a new stage from scratch #######
-FROM alpine:latest  
-
-RUN apk --no-cache add ca-certificates
 
+FROM alpine
 WORKDIR /root/
+COPY --from=build_container /app/configServer .
 
-EXPOSE 8000
-
-# Copy the Pre-built binary file from the previous stage
-COPY --from=builder /app/main .
+EXPOSE 8080
 
-# Command to run the executable
-CMD ["./main"]
+ENTRYPOINT ["./configServer"]

go.mod

@@ -1,21 +1,57 @@
-module github.com/c12s/kuiper
+module kuiper
 
 go 1.18
 
 require (
-	github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da // indirect
-	github.com/fatih/color v1.9.0 // indirect
-	github.com/google/uuid v1.3.0 // indirect
-	github.com/hashicorp/consul/api v1.13.0 // indirect
-	github.com/hashicorp/go-cleanhttp v0.5.1 // indirect
-	github.com/hashicorp/go-hclog v0.12.0 // indirect
-	github.com/hashicorp/go-immutable-radix v1.0.0 // indirect
-	github.com/hashicorp/go-rootcerts v1.0.2 // indirect
-	github.com/hashicorp/golang-lru v0.5.0 // indirect
-	github.com/hashicorp/serf v0.9.6 // indirect
-	github.com/mattn/go-colorable v0.1.6 // indirect
-	github.com/mattn/go-isatty v0.0.12 // indirect
-	github.com/mitchellh/go-homedir v1.1.0 // indirect
-	github.com/mitchellh/mapstructure v1.1.2 // indirect
-	golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44 // indirect
+	github.com/gin-gonic/gin v1.9.0
+	github.com/google/uuid v1.3.0
+	github.com/nats-io/nats.go v1.24.0
+	go.etcd.io/etcd/client/v3 v3.5.7
+	go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.40.0
+	go.opentelemetry.io/otel v1.14.0
+	go.opentelemetry.io/otel/exporters/jaeger v1.14.0
+	go.opentelemetry.io/otel/sdk v1.14.0
+	go.opentelemetry.io/otel/trace v1.14.0
+)
+
+require (
+	github.com/bytedance/sonic v1.8.5 // indirect
+	github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
+	github.com/coreos/go-semver v0.3.0 // indirect
+	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
+	github.com/gin-contrib/sse v0.1.0 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-playground/locales v0.14.1 // indirect
+	github.com/go-playground/universal-translator v0.18.1 // indirect
+	github.com/go-playground/validator/v10 v10.11.2 // indirect
+	github.com/goccy/go-json v0.10.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
+	github.com/leodido/go-urn v1.2.2 // indirect
+	github.com/mattn/go-isatty v0.0.17 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/nats-io/nats-server/v2 v2.9.15 // indirect
+	github.com/nats-io/nkeys v0.3.0 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/pelletier/go-toml/v2 v2.0.7 // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
+	github.com/ugorji/go/codec v1.2.11 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.7 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.5.7 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
+	go.uber.org/zap v1.17.0 // indirect
+	golang.org/x/arch v0.3.0 // indirect
+	golang.org/x/crypto v0.7.0 // indirect
+	golang.org/x/net v0.8.0 // indirect
+	golang.org/x/sys v0.6.0 // indirect
+	golang.org/x/text v0.8.0 // indirect
+	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
+	google.golang.org/grpc v1.53.0 // indirect
+	google.golang.org/protobuf v1.30.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

main.go

@@ -1,42 +1,98 @@
 package main
 
 import (
-	"github.com/c12s/kuiper/model"
-	"github.com/c12s/kuiper/repository/consul"
-	"github.com/c12s/kuiper/service"
+	"context"
+	"kuiper/server"
+	"kuiper/service"
+	"kuiper/store"
+	"kuiper/util"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/propagation"
 )
 
+const kuiperName = "kuiper"
+
 func main() {
-	configRepo, err := consul.New()
+	logger := log.Default()
+	config, err := server.NewConfig()
+	if err != nil {
+		logger.Fatalf("Error: %s", err.Error())
+	}
+
+	ctx := context.Background()
+	//init exporter
+	exporter, err := util.NewJaegerExporter(config.JaegerAddress)
+	if err != nil {
+		logger.Fatalf(err.Error())
+	}
+	//init traceprovider
+	tp := util.NewTraceProvider(exporter)
+	defer func() { _ = tp.Shutdown(ctx) }()
+	otel.SetTracerProvider(tp)
+	tracer := tp.Tracer(kuiperName)
+	otel.SetTextMapPropagator(propagation.TraceContext{})
+
+	router := gin.New()
+	router.Use(gin.Recovery())
+	router.Use(otelgin.Middleware(kuiperName))
 
+	cli, err := clientv3.New(clientv3.Config{
+		Endpoints:   []string{config.EtcdAddress},
+		DialTimeout: 10 * time.Second,
+	})
 	if err != nil {
-		panic(err)
+		logger.Fatalf(err.Error())
 	}
 
-	configService := service.New(configRepo)
-
-	group := model.Group{
-		Configs: []model.Config{
-			model.Config{
-				Key:    "mysql.user",
-				Value:  "asdf",
-				Labels: []model.Label{},
-			},
-			model.Config{
-				Key:    "mysql.pass",
-				Value:  "123",
-				Labels: []model.Label{},
-			},
-		},
+	natsCon, err := util.Conn(config.NatsAddress)
+	if err != nil {
+		logger.Fatalf(err.Error())
 	}
 
-	resp, _ := configService.CreateNewGroup(group)
+	cfgStore := store.NewConfigStore(*cli, *logger, tracer)
+	cfgService := service.NewConfigService(cfgStore, *logger, tracer)
+	handler := server.NewConfigHandler(tracer, *logger, cfgService, *natsCon)
+
+	router.POST("/api/config", handler.SaveConfig)
+	router.GET("/api/config/:id/:ver", handler.GetConfig)
+	router.GET("/api/config/:id", handler.GetConfigsByService)
+	router.POST("/api/config/:id/", handler.CreateNewVersion)
+	router.DELETE("/api/config/:id/:ver", handler.DeleteConfig)
+	router.DELETE("/api/config/:id", handler.DeleteConfigsWithPrefix)
+
+	// start server
+	srv := &http.Server{Addr: "0.0.0.0:8080", Handler: router}
+	go func() {
+		log.Println("server starting")
+		if err := srv.ListenAndServe(); err != nil {
+			if err != http.ErrServerClosed {
+				log.Fatal(err)
+			}
+		}
+	}()
+
+	quit := make(chan os.Signal)
+	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
+	<-quit
 
-	println(resp.Id, resp.Version)
+	log.Println("service shutting down ...")
 
-	stored, _ := configService.GetGroupConfigs(resp.Id, resp.Version, []model.Label{})
+	// gracefully stop server
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 
-	for k, v := range stored {
-		println(k, v)
+	if err := srv.Shutdown(ctx); err != nil {
+		log.Fatal(err)
 	}
-}
+	log.Println("server stopped")
+}

model/config.go

@@ -0,0 +1,9 @@
+package model
+
+type Entries = map[string]string
+
+type Config struct {
+	Service string  `json:"service"`
+	Version string  `json:"version"`
+	Entries Entries `json:"entries"`
+}

server/config.go

@@ -0,0 +1,35 @@
+package server
+
+import (
+	"errors"
+	"os"
+)
+
+type Config struct {
+	JaegerAddress string
+	EtcdAddress   string
+	NatsAddress   string
+}
+
+const (
+	jaegerAddressEnv = "JAEGER_ADDRESS"
+	etcdAddressEnv   = "ETCD_ADDRESS"
+	natsAddressEnv   = "NATS_ADDRESS"
+)
+
+func NewConfig() (Config, error) {
+	jagAddr, found := os.LookupEnv(jaegerAddressEnv)
+	if !found {
+		return Config{}, errors.New(jaegerAddressEnv + " environment variable not set")
+	}
+	etcdAddr, found := os.LookupEnv(etcdAddressEnv)
+	if !found {
+		return Config{}, errors.New(etcdAddressEnv + " environment variable not set")
+	}
+	natsAddr, found := os.LookupEnv(natsAddressEnv)
+	if !found {
+		return Config{}, errors.New(natsAddressEnv + " environment variable not set")
+	}
+
+	return Config{JaegerAddress: jagAddr, EtcdAddress: etcdAddr, NatsAddress: natsAddr}, nil
+}