feat: add repo and processor

Author: pedroduarten9

Makefile

@@ -1,5 +1,11 @@
 install:
 	go install github.com/spf13/cobra-cli@latest
 
+test:
+	go test -race ./...
+
+bench:
+	go test ./... -bench=.
+
 serve:
 	go run main.go serve

docs/decisions.md

@@ -6,4 +6,20 @@ This document will outline the technical and product decisions taken along the c
 
 ### Usage of Cobra
 
-Cobra CLI is a CLI helper that makes helps with the running of the service, so I opted to use it.
+Cobra CLI is a CLI helper that makes helps with the running of the service, so I opted to use it.
+
+### Usage of mocks
+
+There are interfaces in case I wanted to mock requests, I saw no need for that since I can easily test the full flow without depending on external sources or flaky I/O. Therefore I decided not to use mocks.
+
+### Support for log values
+
+I opted to support multiple value types for the logs but I will only test for strings, otherwise it would take me much more time. I can also remove the support.
+
+### Log processing response
+
+For this iteration I can either find or not the key, I will not be processing errors, therefore the return is always the same.
+
+### Concurrency
+
+There are unused structs such as concurrent repo and concurrent service that use a synchronized map with sharding and go routines respectively, with the benchmarks result the decision was to keep the simpler versions.

go.mod

@@ -3,13 +3,19 @@ module pedroduarten9/oltp-log-processor
 go 1.23.0
 
 require (
+	github.com/orcaman/concurrent-map/v2 v2.0.1
 	github.com/spf13/cobra v1.10.1
+	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/proto/otlp v1.8.0
 )
 
 require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/spf13/pflag v1.0.9 // indirect
 	golang.org/x/net v0.43.0 // indirect
 	golang.org/x/sys v0.35.0 // indirect
@@ -18,4 +24,5 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 // indirect
 	google.golang.org/grpc v1.75.0 // indirect
 	google.golang.org/protobuf v1.36.8 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,4 +1,7 @@
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
 github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
@@ -13,11 +16,23 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnV
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/orcaman/concurrent-map/v2 v2.0.1 h1:jOJ5Pg2w1oeB6PeDurIYf6k9PQ+aTITr/6lP/L/zp6c=
+github.com/orcaman/concurrent-map/v2 v2.0.1/go.mod h1:9Eq3TG2oBe5FirmYWQfYO5iH1q0Jv47PLaNK++uCdOM=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
+github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/spf13/cobra v1.10.1 h1:lJeBwCfmrnXthfAupyUTzJ/J4Nc1RsHC/mSRU2dll/s=
 github.com/spf13/cobra v1.10.1/go.mod h1:7SmJGaTHFVBY0jW4NXGluQoLvhqFQM+6XSKD+P4XaB0=
 github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY=
 github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
 go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
@@ -49,4 +64,7 @@ google.golang.org/grpc v1.75.0/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr
 google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
 google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/logs/concurrent_repo.go

@@ -0,0 +1,24 @@
+package logs
+
+import cmap "github.com/orcaman/concurrent-map/v2"
+
+type concurrentRepo struct {
+	counts cmap.ConcurrentMap[string, int]
+}
+
+func NewConcurrentRepo() *concurrentRepo {
+	return &concurrentRepo{
+		counts: cmap.New[int](),
+	}
+}
+
+func (r *concurrentRepo) IncrementAttribute(attr string) {
+	val, _ := r.counts.Get(attr)
+	r.counts.Set(attr, val+1)
+}
+
+func (r *concurrentRepo) Reset() map[string]int {
+	oldMap := r.counts
+	r.counts = cmap.New[int]()
+	return oldMap.Items()
+}

pkg/logs/concurrent_repo_test.go

@@ -0,0 +1,109 @@
+package logs
+
+import (
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConcurrentIncrementAttributeAndReset(t *testing.T) {
+	repo := NewConcurrentRepo()
+
+	tests := []struct {
+		name               string
+		attributesAndCount map[string]int
+	}{
+		{
+			name:               "single key",
+			attributesAndCount: map[string]int{"key": 1},
+		},
+		{
+			name:               "single key multiple times",
+			attributesAndCount: map[string]int{"key": 3},
+		},
+		{
+			name:               "multiple keys",
+			attributesAndCount: map[string]int{"key": 3, "key2": 1, "key3": 4},
+		},
+	}
+
+	for _, tt := range tests {
+		for k, v := range tt.attributesAndCount {
+			for i := 0; i < v; i++ {
+				repo.IncrementAttribute(k)
+			}
+		}
+
+		oldMap := repo.Reset()
+		assert.Equal(t, tt.attributesAndCount, oldMap)
+		assert.True(t, repo.counts.IsEmpty())
+	}
+}
+
+func BenchmarkConcurrentRepo_Serial(b *testing.B) {
+	key := "foo"
+	repo := NewConcurrentRepo()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		repo.IncrementAttribute(key)
+	}
+}
+
+func BenchmarkConcurrentRepo_Parallel(b *testing.B) {
+	key := "foo"
+	repo := NewConcurrentRepo()
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			repo.IncrementAttribute(key)
+		}
+	})
+}
+
+func BenchmarkConcurrentRepo_ManyKeys(b *testing.B) {
+	repo := NewConcurrentRepo()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		val := strconv.Itoa(i % 1000)
+		repo.IncrementAttribute(val)
+	}
+}
+
+func BenchmarkConcurrentRepo_WriteAndReset(b *testing.B) {
+	repo := NewConcurrentRepo()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		val := strconv.Itoa(i % 1000)
+		repo.IncrementAttribute(val)
+	}
+
+	for i := 0; i < b.N; i++ {
+		_ = repo.Reset()
+	}
+}
+
+func BenchmarkConcurrentRepo_ConcurrentWriteAndRead(b *testing.B) {
+	key := "foo"
+	repo := NewConcurrentRepo()
+
+	b.ResetTimer()
+	var wg sync.WaitGroup
+	for i := 0; i < b.N; i++ {
+		wg.Add(2)
+		go func() {
+			repo.IncrementAttribute(key)
+			wg.Done()
+		}()
+		go func() {
+			_ = repo.Reset()
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+}

pkg/logs/concurrent_service.go

@@ -0,0 +1,30 @@
+package logs
+
+import (
+	"context"
+	"log/slog"
+
+	collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+)
+
+type dash0LogsConcurrentServiceServer struct {
+	logProcessor *LogProcessor
+
+	collogspb.UnimplementedLogsServiceServer
+}
+
+func NewConcurrentServer(logProcessor *LogProcessor) collogspb.LogsServiceServer {
+	return &dash0LogsConcurrentServiceServer{
+		logProcessor: logProcessor,
+	}
+}
+
+func (l *dash0LogsConcurrentServiceServer) Export(ctx context.Context, request *collogspb.ExportLogsServiceRequest) (*collogspb.ExportLogsServiceResponse, error) {
+	slog.DebugContext(ctx, "Received ExportLogsServiceRequest")
+
+	for _, resourceLogs := range request.ResourceLogs {
+		go l.logProcessor.ProcessLog(resourceLogs)
+	}
+
+	return &collogspb.ExportLogsServiceResponse{}, nil
+}

pkg/logs/concurrent_service_test.go

@@ -0,0 +1,17 @@
+package logs
+
+import (
+	"context"
+	"testing"
+)
+
+func BenchmarkConcurrentExport_9kBatch(b *testing.B) {
+	key := "foo"
+	ctx := context.Background()
+	service := NewConcurrentServer(NewLogProcessor(key, NewRepo()))
+
+	for i := 0; i < b.N; i++ {
+		logs := generateLog(key, 9000)
+		service.Export(ctx, logs)
+	}
+}