feat: add gRPC server

Author: pedroduarten9

Dockerfile

@@ -14,6 +14,7 @@ FROM gcr.io/distroless/base
 WORKDIR /
 
 COPY --from=build-stage --chown=serve:serve /app/serve .
+COPY --from=build-stage --chown=serve:serve /app/config.yaml .
 
 USER nonroot:nonroot
 

Makefile

@@ -1,4 +1,5 @@
 install:
+	brew install grpcurl
 	go install github.com/spf13/cobra-cli@latest
 
 test:

README.md

@@ -11,6 +11,18 @@ It encompasses documentation that helps extend this service and facilitates its
 
 The decisions taken along the challenge will be documented [here](./docs/decisions.md)
 
+## Flags
+
+The configurable attributes `window in seconds` and `attribute key` can be configured either by changing the values on the `config.yaml` file or by passing flags via command line to the executable.
+Flags:
+- attrKey
+- windowSeconds
+Example:
+`go run main.go serve --attrKey=foo --windowSeconds=7`
+
+The same command can be inferred either to be run via docker or via shell script.
+There are other flags that can be checked on `config.yaml` file
+
 ## How to run (via Golang)
 
 To run the service you can either use Golang directly or use the helper Makefile, the commands for each of the approaches are below.
@@ -41,18 +53,26 @@ Have Docker installed
 
 ## How to run (via shell scripts)
 
-## How to start the processor
+### How to start the processor
 
 In order to start the server a helper file was created `run.sh`.runs the container with the application, when finished it removes the container.  
 
-## How to check the logs
+### How to check the logs
 
 In order to check the logs of the application one can run `inspect.sh`. Once executed it will be continuously listen to the logs on the container.
 
-## How to delete the resources created
+### How to delete the resources created
 
 In order to delete the resources created a helper file was created `cleanup.sh`. This executable deletes the image.
 
 ### Prerequisites
 
-Have Docker installed
+Have Docker installed
+
+## Exercise the application
+
+To exercise the application just run the server with the flags you wish.
+To run a request make sure you have brew installed and run `make install` to install a gRPC client (make sure you have brew installed).
+Otherwise, please, just use another client.
+Command example: `grpcurl -plaintext -d @ localhost:4317 opentelemetry.proto.collector.logs.v1.LogsService/Export < examples/log.json`
+Where you can change the address if it's different and provide different files.

cleanup.sh

@@ -1,5 +1,4 @@
 #!/bin/bash
-set -e
-
+docker stop log-processor
 docker image rm log-processor
 echo "Docker image 'log-processor' removed successfully."

cmd/root.go

@@ -1,34 +1,24 @@
-/*
-Copyright © 2025 NAME HERE <EMAIL ADDRESS>
-
-*/
 package cmd
 
 import (
+	"errors"
 	"os"
+	"strings"
 
 	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
 )
 
+var cfgFile string
 
-
-// rootCmd represents the base command when called without any subcommands
 var rootCmd = &cobra.Command{
 	Use:   "oltp-log-processor",
-	Short: "A brief description of your application",
-	Long: `A longer description that spans multiple lines and likely contains
-examples and usage of using your application. For example:
-
-Cobra is a CLI library for Go that empowers applications.
-This application is a tool to generate the needed files
-to quickly create a Cobra application.`,
-	// Uncomment the following line if your bare application
-	// has an action associated with it:
-	// Run: func(cmd *cobra.Command, args []string) { },
+	Short: "OLTP log processor",
+	PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
+		return initializeConfig(cmd)
+	},
 }
 
-// Execute adds all child commands to the root command and sets flags appropriately.
-// This is called by main.main(). It only needs to happen once to the rootCmd.
 func Execute() {
 	err := rootCmd.Execute()
 	if err != nil {
@@ -37,15 +27,36 @@ func Execute() {
 }
 
 func init() {
-	// Here you will define your flags and configuration settings.
-	// Cobra supports persistent flags, which, if defined here,
-	// will be global for your application.
+	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default locations: ., $HOME/.myapp/)")
+}
 
-	// rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is $HOME/.oltp-log-processor.yaml)")
+func initializeConfig(cmd *cobra.Command) error {
+	viper.SetEnvKeyReplacer(strings.NewReplacer(".", "*", "-", "*"))
+	viper.AutomaticEnv()
 
-	// Cobra also supports local flags, which will only run
-	// when this action is called directly.
-	rootCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
-}
+	if cfgFile != "" {
+		viper.SetConfigFile(cfgFile)
+	} else {
+		home, err := os.UserHomeDir()
+		cobra.CheckErr(err)
 
+		viper.AddConfigPath(".")
+		viper.AddConfigPath(home)
+		viper.SetConfigName("config")
+		viper.SetConfigType("yaml")
+	}
 
+	if err := viper.ReadInConfig(); err != nil {
+		var configFileNotFoundError viper.ConfigFileNotFoundError
+		if !errors.As(err, &configFileNotFoundError) {
+			return err
+		}
+	}
+
+	err := viper.BindPFlags(cmd.Flags())
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

cmd/serve.go

@@ -1,40 +1,107 @@
-/*
-Copyright © 2025 NAME HERE <EMAIL ADDRESS>
-
-*/
 package cmd
 
 import (
+	"context"
 	"fmt"
+	"log/slog"
+	"net"
+	"os"
+	"os/signal"
+	"pedroduarten9/oltp-log-processor/pkg/logs"
+	"time"
 
 	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/reflection"
 )
 
-// serveCmd represents the serve command
 var serveCmd = &cobra.Command{
 	Use:   "serve",
-	Short: "A brief description of your command",
-	Long: `A longer description that spans multiple lines and likely contains examples
-and usage of using your command. For example:
-
-Cobra is a CLI library for Go that empowers applications.
-This application is a tool to generate the needed files
-to quickly create a Cobra application.`,
+	Short: "A grpc server for the log service",
 	Run: func(cmd *cobra.Command, args []string) {
-		fmt.Println("serve called")
+		windowSeconds := viper.GetInt("windowSeconds")
+		maxReceiveMessageSize := viper.GetInt("maxReceiveMessageSize")
+		attrKey := viper.GetString("attrKey")
+		listenAddr := viper.GetString("listenAddr")
+		logLevelCfg := viper.GetString("logLevel")
+
+		logLevel := *new(slog.Level)
+		if err := logLevel.UnmarshalText([]byte(logLevelCfg)); err != nil {
+			logLevel = slog.LevelInfo
+		}
+		logger := slog.New(slog.NewJSONHandler(
+			os.Stdout,
+			&slog.HandlerOptions{Level: logLevel}))
+		slog.SetDefault(logger)
+
+		repo := logs.NewRepo()
+		logProcessor := logs.NewLogProcessor(attrKey, repo)
+		ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, os.Kill)
+		start(ctx, logProcessor, logger, windowSeconds)
+
+		slog.Debug("Starting listener", slog.String("listenAddr", listenAddr))
+		listener, err := net.Listen("tcp", listenAddr)
+		if err != nil {
+			slog.Error("failed to create listener")
+			return
+		}
+
+		grpcServer := grpc.NewServer(
+			grpc.StatsHandler(otelgrpc.NewServerHandler()),
+			grpc.MaxRecvMsgSize(maxReceiveMessageSize),
+			grpc.Creds(insecure.NewCredentials()),
+		)
+		collogspb.RegisterLogsServiceServer(grpcServer, logs.NewServer(logProcessor))
+
+		reflection.Register(grpcServer)
+		slog.Debug("Starting gRPC server")
+
+		go func() {
+			if err := grpcServer.Serve(listener); err != nil {
+				cancel()
+			}
+		}()
+
+		<-ctx.Done()
+
+		slog.Debug("Stopping gRPC server")
+		grpcServer.GracefulStop()
+
+		_ = listener.Close()
 	},
 }
 
 func init() {
 	rootCmd.AddCommand(serveCmd)
+	serveCmd.Flags().Int("windowSeconds", 30, "Window of logging of attributes in seconds")
+	serveCmd.Flags().Int("maxReceiveMessageSize", 16777216, "Max receive message size for the gRPC handler")
+	serveCmd.Flags().String("attrKey", "foo", "The attribute key to look for")
+	serveCmd.Flags().String("listenAddr", "localhost:4317", "The listen address for the gRPC server")
+	serveCmd.Flags().String("logLevel", "INFO", "The log level of the system")
+}
 
-	// Here you will define your flags and configuration settings.
-
-	// Cobra supports Persistent Flags which will work for this command
-	// and all subcommands, e.g.:
-	// serveCmd.PersistentFlags().String("foo", "", "A help for foo")
-
-	// Cobra supports local flags which will only run when this command
-	// is called directly, e.g.:
-	// serveCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
+func start(ctx context.Context, lp *logs.LogProcessor, logger *slog.Logger, windowSeconds int) {
+	ticker := time.NewTicker(time.Duration(windowSeconds) * time.Second)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				report := lp.ReportAndReset()
+				if len(report) == 0 {
+					logger.Info(fmt.Sprintf("No logs found for window (%d seconds).", windowSeconds))
+					break
+				}
+				intro := fmt.Sprintf("Reporting counts for window (%d seconds):", windowSeconds)
+				logger.Info(intro)
+				logger.Info(report)
+			case <-ctx.Done():
+				ticker.Stop()
+				return
+			}
+		}
+	}()
 }

cmd/test/serve_test.go

@@ -0,0 +1,100 @@
+package test
+
+import (
+	"context"
+	"log"
+	"net"
+	"pedroduarten9/oltp-log-processor/pkg/logs"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+	otellogs "go.opentelemetry.io/proto/otlp/logs/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+func TestLogsServiceServer_Export(t *testing.T) {
+	ctx := context.Background()
+
+	client, closer := server()
+	defer closer()
+
+	type expectation struct {
+		out *collogspb.ExportLogsServiceResponse
+		err error
+	}
+
+	tests := map[string]struct {
+		in       *collogspb.ExportLogsServiceRequest
+		expected expectation
+	}{
+		"Must_Success": {
+			in: &collogspb.ExportLogsServiceRequest{
+				ResourceLogs: []*otellogs.ResourceLogs{
+					{
+						ScopeLogs: []*otellogs.ScopeLogs{},
+						SchemaUrl: "dash0.com/otlp-log-processor-backend",
+					},
+				},
+			},
+			expected: expectation{
+				out: &collogspb.ExportLogsServiceResponse{},
+				err: nil,
+			},
+		},
+	}
+
+	for scenario, tt := range tests {
+		t.Run(scenario, func(t *testing.T) {
+			out, err := client.Export(ctx, tt.in)
+			if err != nil {
+				assert.Equal(t, tt.expected.err, err)
+			} else {
+				expectedPartialSuccess := tt.expected.out.GetPartialSuccess()
+				partialSuccess := out.GetPartialSuccess()
+				assert.Equal(t, expectedPartialSuccess.GetRejectedLogRecords(), partialSuccess.GetRejectedLogRecords())
+				assert.Equal(t, expectedPartialSuccess.GetErrorMessage(), partialSuccess.GetErrorMessage())
+			}
+
+		})
+	}
+}
+
+func server() (collogspb.LogsServiceClient, func()) {
+	addr := "localhost:4317"
+	buffer := 101024 * 1024
+	lis := bufconn.Listen(buffer)
+
+	var err error
+
+	baseServer := grpc.NewServer()
+	logsProcessor := logs.NewLogProcessor("foo", logs.NewRepo())
+	collogspb.RegisterLogsServiceServer(baseServer, logs.NewServer(logsProcessor))
+	go func() {
+		if err := baseServer.Serve(lis); err != nil {
+			log.Printf("error serving server: %v", err)
+		}
+	}()
+
+	conn, err := grpc.NewClient(addr,
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		log.Printf("error connecting to server: %v", err)
+	}
+
+	closer := func() {
+		err := lis.Close()
+		if err != nil {
+			log.Printf("error closing listener: %v", err)
+		}
+		baseServer.Stop()
+	}
+
+	client := collogspb.NewLogsServiceClient(conn)
+
+	return client, closer
+}

config.yaml

@@ -0,0 +1,5 @@
+listenAddr: "localhost:4317"
+maxReceiveMessageSize: 16777216
+windowSeconds: 5
+attrKey: "foo"
+logLevel: "DEBUG"