Feat: Raccoon CLI

.github/workflows/test.yaml

@@ -47,6 +47,8 @@ jobs:
       - uses: actions/checkout@v2
       - name: Build
         run: make build
+      - name: Smoke test
+        run: ./raccoon
   benchmark:
     runs-on: ubuntu-latest
     steps:

Dockerfile

@@ -14,4 +14,4 @@ RUN make build
 FROM debian:bookworm-slim
 WORKDIR /app
 COPY --from=0 /app/raccoon ./raccoon
-CMD ["./raccoon"]
+CMD ["./raccoon", "server"]

Dockerfile.release

@@ -1,4 +1,4 @@
 FROM debian:bookworm-slim
 WORKDIR /app
 COPY raccoon .
-CMD ["./raccoon"]
+CMD ["./raccoon", "server"]

app/server.go

@@ -35,20 +35,20 @@ type Publisher interface {
 
 // StartServer starts the server
 func StartServer(ctx context.Context, cancel context.CancelFunc) {
-	bufferChannel := make(chan collector.CollectRequest, config.Worker.ChannelSize)
+	bufferChannel := make(chan collector.CollectRequest, config.Server.Worker.ChannelSize)
 	httpServices := services.Create(bufferChannel)
 	logger.Info("Start Server -->")
 	httpServices.Start(ctx, cancel)
-	logger.Infof("Start publisher --> %s", config.Publisher)
+	logger.Infof("Start publisher --> %s", config.Server.Publisher)
 	publisher, err := initPublisher()
 	if err != nil {
-		logger.Errorf("Error creating %q publisher: %v\n", config.Publisher, err)
+		logger.Errorf("Error creating %q publisher: %v\n", config.Server.Publisher, err)
 		logger.Info("Exiting server")
 		os.Exit(0)
 	}
 
 	logger.Info("Start worker -->")
-	workerPool := worker.CreateWorkerPool(config.Worker.WorkersPoolSize, bufferChannel, config.Worker.DeliveryChannelSize, publisher)
+	workerPool := worker.CreateWorkerPool(config.Server.Worker.WorkersPoolSize, bufferChannel, config.Server.Worker.DeliveryChannelSize, publisher)
 	workerPool.StartWorkers()
 
 	go reportProcMetrics()
@@ -58,18 +58,19 @@ func StartServer(ctx context.Context, cancel context.CancelFunc) {
 func shutDownServer(ctx context.Context, cancel context.CancelFunc, httpServices services.Services, bufferChannel chan collector.CollectRequest, workerPool *worker.Pool, pub Publisher) {
 	signalChan := make(chan os.Signal, 1)
 	signal.Notify(signalChan, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	workerFlushTimeout := time.Duration(config.Server.Worker.WorkerFlushTimeoutMS) * time.Millisecond
 	for {
 		sig := <-signalChan
 		switch sig {
 		case syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT:
 			logger.Info(fmt.Sprintf("[App.Server] Received a signal %s", sig))
 			httpServices.Shutdown(ctx)
 			logger.Info("Server shutdown all the listeners")
-			timedOut := workerPool.FlushWithTimeOut(config.Worker.WorkerFlushTimeout)
+			timedOut := workerPool.FlushWithTimeOut(workerFlushTimeout)
 			if timedOut {
 				logger.Info(fmt.Sprintf("WorkerPool flush timedout %t", timedOut))
 			}
-			flushInterval := config.PublisherKafka.FlushInterval
+			flushInterval := config.Server.PublisherKafka.FlushInterval
 			logger.Infof("Closing %q producer\n", pub.Name())
 			logger.Info(fmt.Sprintf("Wait %d ms for all messages to be delivered", flushInterval))
 			eventsInProducer := 0
@@ -108,7 +109,8 @@ func shutDownServer(ctx context.Context, cancel context.CancelFunc, httpServices
 
 func reportProcMetrics() {
 	m := &runtime.MemStats{}
-	for range time.Tick(config.MetricInfo.RuntimeStatsRecordInterval) {
+	reportInterval := time.Duration(config.Server.MetricInfo.RuntimeStatsRecordIntervalMS) * time.Millisecond
+	for range time.Tick(reportInterval) {
 		metrics.Gauge("server_go_routines_count_current", runtime.NumGoroutine(), map[string]string{})
 		runtime.ReadMemStats(m)
 		metrics.Gauge("server_mem_heap_alloc_bytes_current", m.HeapAlloc, map[string]string{})
@@ -123,50 +125,60 @@ func reportProcMetrics() {
 }
 
 func initPublisher() (Publisher, error) {
-	switch config.Publisher {
+	switch config.Server.Publisher {
 	case "kafka":
 		return kafka.New()
 	case "pubsub":
 		client, err := pubsubsdk.NewClient(
 			context.Background(),
-			config.PublisherPubSub.ProjectId,
-			option.WithCredentialsFile(config.PublisherPubSub.CredentialsFile),
+			config.Server.PublisherPubSub.ProjectId,
+			option.WithCredentialsFile(config.Server.PublisherPubSub.CredentialsFile),
 		)
 		if err != nil {
 			return nil, fmt.Errorf("error creating pubsub client: %w", err)
 		}
+		var (
+			topicRetention = time.Duration(config.Server.PublisherPubSub.TopicRetentionPeriodMS) * time.Millisecond
+			delayThreshold = time.Duration(config.Server.PublisherPubSub.PublishDelayThresholdMS) * time.Millisecond
+			publishTimeout = time.Duration(config.Server.PublisherPubSub.PublishTimeoutMS) * time.Millisecond
+		)
 		return pubsub.New(
 			client,
-			pubsub.WithTopicFormat(config.EventDistribution.PublisherPattern),
-			pubsub.WithTopicAutocreate(config.PublisherPubSub.TopicAutoCreate),
-			pubsub.WithTopicRetention(config.PublisherPubSub.TopicRetentionPeriod),
-			pubsub.WithDelayThreshold(config.PublisherPubSub.PublishDelayThreshold),
-			pubsub.WithCountThreshold(config.PublisherPubSub.PublishCountThreshold),
-			pubsub.WithByteThreshold(config.PublisherPubSub.PublishByteThreshold),
-			pubsub.WithTimeout(config.PublisherPubSub.PublishTimeout),
+			pubsub.WithTopicFormat(config.Server.EventDistribution.PublisherPattern),
+			pubsub.WithTopicAutocreate(config.Server.PublisherPubSub.TopicAutoCreate),
+			pubsub.WithTopicRetention(topicRetention),
+			pubsub.WithDelayThreshold(delayThreshold),
+			pubsub.WithCountThreshold(config.Server.PublisherPubSub.PublishCountThreshold),
+			pubsub.WithByteThreshold(config.Server.PublisherPubSub.PublishByteThreshold),
+			pubsub.WithTimeout(publishTimeout),
 		)
 	case "kinesis":
 		cfg, err := awsconfig.LoadDefaultConfig(
 			context.Background(),
-			awsconfig.WithRegion(config.PublisherKinesis.Region),
+			awsconfig.WithRegion(config.Server.PublisherKinesis.Region),
 			awsconfig.WithSharedConfigFiles(
-				[]string{config.PublisherKinesis.CredentialsFile},
+				[]string{config.Server.PublisherKinesis.CredentialsFile},
 			),
 		)
 		if err != nil {
 			return nil, fmt.Errorf("error locating aws credentials: %w", err)
 		}
-		conf := config.PublisherKinesis
+		var (
+			conf           = config.Server.PublisherKinesis
+			publishTimeout = time.Duration(conf.PublishTimeoutMS) * time.Millisecond
+			probeInterval  = time.Duration(conf.StreamProbeIntervalMS) * time.Millisecond
+		)
+
 		return kinesis.New(
 			kinesissdk.NewFromConfig(cfg),
-			kinesis.WithStreamPattern(config.EventDistribution.PublisherPattern),
+			kinesis.WithStreamPattern(config.Server.EventDistribution.PublisherPattern),
 			kinesis.WithStreamAutocreate(conf.StreamAutoCreate),
 			kinesis.WithStreamMode(types.StreamMode(conf.StreamMode)),
 			kinesis.WithShards(conf.DefaultShards),
-			kinesis.WithPublishTimeout(conf.PublishTimeout),
-			kinesis.WithStreamProbleInterval(conf.StreamProbeInterval),
+			kinesis.WithPublishTimeout(publishTimeout),
+			kinesis.WithStreamProbleInterval(probeInterval),
 		)
 	default:
-		return nil, fmt.Errorf("unknown publisher: %v", config.Publisher)
+		return nil, fmt.Errorf("unknown publisher: %v", config.Server.Publisher)
 	}
 }

cmd/root.go

@@ -0,0 +1,29 @@
+package cmd
+
+import (
+	"github.com/MakeNowJust/heredoc"
+	"github.com/spf13/cobra"
+)
+
+func New() *cobra.Command {
+	root := &cobra.Command{
+		Use:          "raccoon",
+		Short:        "Scalable event ingestion tool",
+		SilenceUsage: true,
+		CompletionOptions: cobra.CompletionOptions{
+			DisableDefaultCmd: true,
+		},
+		Args: cobra.NoArgs,
+		Long: heredoc.Doc(`
+			Raccoon is a high-throughput, low-latency service to collect 
+			events in real-time from your web, mobile apps, and services 
+			using multiple network protocols.`),
+		Run: func(cmd *cobra.Command, args []string) {
+			cmd.Help()
+		},
+	}
+
+	root.AddCommand(serverCommand())
+	root.SetHelpCommand(&cobra.Command{Hidden: true})
+	return root
+}