diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 81c4dfd..85d9fea 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,3 +30,10 @@ jobs: - name: Run Integration Test run: go test -v -tags=integration + + - name: Build Docker Images + run: | + docker build -t ingestor -f ingestor/Dockerfile . + docker build -t parser -f parser/Dockerfile . + docker build -t storage-writer -f storage-writer/Dockerfile . + docker build -t log-generator -f log-generator/Dockerfile . diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..13f2bf2 --- /dev/null +++ b/config.yaml @@ -0,0 +1,20 @@ +kafka: + brokers: ["kafka:9092"] + topics: + raw: "raw_logs" + parsed: "parsed_logs" + raw_dlq: "raw_logs_dlq" + parsed_dlq: "parsed_logs_dlq" + +elasticsearch: + url: "http://elasticsearch:9200" + +ingestor: + http_port: 8081 + +parser: + version: "1.0.0" + +storage_writer: + version: "1.0.0" + index_prefix: "logs" \ No newline at end of file diff --git a/config/config.go b/config/config.go new file mode 100644 index 0000000..579ba83 --- /dev/null +++ b/config/config.go @@ -0,0 +1,59 @@ +package config + +import ( + "github.com/spf13/viper" +) + +type Config struct { + Kafka KafkaConfig + Elasticsearch ElasticsearchConfig + Ingestor IngestorConfig + Parser ParserConfig + StorageWriter StorageWriterConfig `mapstructure:"storage_writer"` +} + +type KafkaConfig struct { + Brokers []string + Topics KafkaTopics +} + +type KafkaTopics struct { + Raw string + Parsed string + RawDLQ string `mapstructure:"raw_dlq"` + ParsedDLQ string `mapstructure:"parsed_dlq"` +} + +type ElasticsearchConfig struct { + URL string +} + +type IngestorConfig struct { + HTTPPort int `mapstructure:"http_port"` +} + +type ParserConfig struct { + Version string `mapstructure:"version"` +} + +type StorageWriterConfig struct { + Version string `mapstructure:"version"` + IndexPrefix string `mapstructure:"index_prefix"` +} + +func LoadConfig() (*Config, error) { + viper.SetConfigName("config") + viper.SetConfigType("yaml") + viper.AddConfigPath(".") + + if err := viper.ReadInConfig(); err != nil { + return nil, err + } + + var config Config + if err := viper.Unmarshal(&config); err != nil { + return nil, err + } + + return &config, nil +} diff --git a/docker-compose.yml b/docker-compose.yml index a083ece..ab2fe64 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -61,6 +61,16 @@ services: environment: - KAFKA_BROKERS=kafka:9092 + log-generator: + build: + context: . + dockerfile: log-generator/Dockerfile + depends_on: + - ingestor + environment: + - INGESTOR_URL=http://ingestor:8081/log + restart: on-failure + parser: build: context: . diff --git a/go.mod b/go.mod index 88abff0..30c0e32 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/IBM/sarama v1.45.2 github.com/elastic/go-elasticsearch/v8 v8.18.0 github.com/gin-gonic/gin v1.10.1 + github.com/spf13/viper v1.20.1 ) require ( @@ -17,6 +18,7 @@ require ( github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect github.com/eapache/queue v1.1.0 // indirect github.com/elastic/elastic-transport-go/v8 v8.7.0 // indirect + github.com/fsnotify/fsnotify v1.8.0 // indirect github.com/gabriel-vasile/mimetype v1.4.9 // indirect github.com/gin-contrib/sse v1.1.0 // indirect github.com/go-logr/logr v1.4.2 // indirect @@ -24,6 +26,7 @@ require ( 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.26.0 // indirect + github.com/go-viper/mapstructure/v2 v2.2.1 // indirect github.com/goccy/go-json v0.10.5 // indirect github.com/golang/snappy v1.0.0 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect @@ -37,7 +40,6 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/compress v1.18.0 // indirect github.com/klauspost/cpuid/v2 v2.2.10 // indirect - github.com/kr/text v0.2.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect @@ -46,11 +48,19 @@ require ( github.com/pierrec/lz4/v4 v4.1.22 // indirect github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect github.com/rogpeppe/go-internal v1.14.1 // indirect + github.com/sagikazarmark/locafero v0.7.0 // indirect + github.com/sourcegraph/conc v0.3.0 // indirect + github.com/spf13/afero v1.12.0 // indirect + github.com/spf13/cast v1.7.1 // indirect + github.com/spf13/pflag v1.0.6 // indirect + github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.14 // indirect - go.opentelemetry.io/otel v1.28.0 // indirect - go.opentelemetry.io/otel/metric v1.28.0 // indirect - go.opentelemetry.io/otel/trace v1.28.0 // indirect + go.opentelemetry.io/otel v1.29.0 // indirect + go.opentelemetry.io/otel/metric v1.29.0 // indirect + go.opentelemetry.io/otel/trace v1.29.0 // indirect + go.uber.org/atomic v1.9.0 // indirect + go.uber.org/multierr v1.9.0 // indirect golang.org/x/arch v0.18.0 // indirect golang.org/x/crypto v0.39.0 // indirect golang.org/x/net v0.41.0 // indirect diff --git a/go.sum b/go.sum index 188215f..c7136f6 100644 --- a/go.sum +++ b/go.sum @@ -8,7 +8,6 @@ github.com/bytedance/sonic/loader v0.2.4/go.mod h1:N8A3vUdtUebEY2/VQC0MyhYeKUFos github.com/cloudwego/base64x v0.1.5 h1:XPciSp1xaq2VCSt6lF0phncD4koWyULpl5bUxbfCyP4= github.com/cloudwego/base64x v0.1.5/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJgA0rcu/8w= github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY= -github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= 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= @@ -24,6 +23,10 @@ github.com/elastic/go-elasticsearch/v8 v8.18.0 h1:ANNq1h7DEiPUaALb8+5w3baQzaS08W github.com/elastic/go-elasticsearch/v8 v8.18.0/go.mod h1:WLqwXsJmQoYkoA9JBFeEwPkQhCfAZuUvfpdU/NvSSf0= github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw= github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= +github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= +github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= +github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M= +github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/gabriel-vasile/mimetype v1.4.9 h1:5k+WDwEsD9eTLL8Tz3L0VnmVh9QxGjRmjBvAG7U/oYY= github.com/gabriel-vasile/mimetype v1.4.9/go.mod h1:WnSQhFKJuBlRyLiKohA/2DtIlPFAbguNaG7QCHcyGok= github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w= @@ -43,6 +46,8 @@ github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJn github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= github.com/go-playground/validator/v10 v10.26.0 h1:SP05Nqhjcvz81uJaRfEV0YBSSSGMc/iMaVtFbr3Sw2k= github.com/go-playground/validator/v10 v10.26.0/go.mod h1:I5QpIEbmr8On7W0TktmJAumgzX4CA1XNl4ZmDuVHKKo= +github.com/go-viper/mapstructure/v2 v2.2.1 h1:ZAaOCxANMuZx5RCeg0mBdEZk7DZasvvZIxtHqx8aGss= +github.com/go-viper/mapstructure/v2 v2.2.1/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs= @@ -50,6 +55,8 @@ github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEW github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4= github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -80,8 +87,8 @@ github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa02 github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/knz/go-libedit v1.10.1/go.mod h1:MZTVkCWyz0oBc7JOWP3wNAzd002ZbM/5hgShxwh4x8M= -github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= -github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= +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/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= @@ -103,6 +110,18 @@ github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8A github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= 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/sagikazarmark/locafero v0.7.0 h1:5MqpDsTGNDhY8sGp0Aowyf0qKsPrhewaLSsFaodPcyo= +github.com/sagikazarmark/locafero v0.7.0/go.mod h1:2za3Cg5rMaTMoG/2Ulr9AwtFaIppKXTRYnozin4aB5k= +github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo= +github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0= +github.com/spf13/afero v1.12.0 h1:UcOPyRBYczmFn6yvphxkn9ZEOY65cpwGKb5mL36mrqs= +github.com/spf13/afero v1.12.0/go.mod h1:ZTlWwG4/ahT8W7T0WQ5uYmjI9duaLQGy3Q2OAl4sk/4= +github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y= +github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= +github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= +github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/viper v1.20.1 h1:ZMi+z/lvLyPSCoNtFCpqjy0S4kPbirhpTMwl8BkW9X4= +github.com/spf13/viper v1.20.1/go.mod h1:P9Mdzt1zoHIG8m2eZQinpiBjo6kCmZSKBClNNqjJvu4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= @@ -114,19 +133,25 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= +github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.14 h1:yOQvXCBc3Ij46LRkRoh4Yd5qK6LVOgi0bYOXfb7ifjw= github.com/ugorji/go/codec v1.2.14/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= -go.opentelemetry.io/otel v1.28.0 h1:/SqNcYk+idO0CxKEUOtKQClMK/MimZihKYMruSMViUo= -go.opentelemetry.io/otel v1.28.0/go.mod h1:q68ijF8Fc8CnMHKyzqL6akLO46ePnjkgfIMIjUIX9z4= -go.opentelemetry.io/otel/metric v1.28.0 h1:f0HGvSl1KRAU1DLgLGFjrwVyismPlnuU6JD6bOeuA5Q= -go.opentelemetry.io/otel/metric v1.28.0/go.mod h1:Fb1eVBFZmLVTMb6PPohq3TO9IIhUisDsbJoL/+uQW4s= -go.opentelemetry.io/otel/sdk v1.21.0 h1:FTt8qirL1EysG6sTQRZ5TokkU8d0ugCj8htOgThZXQ8= -go.opentelemetry.io/otel/sdk v1.21.0/go.mod h1:Nna6Yv7PWTdgJHVRD9hIYywQBRx7pbox6nwBnZIxl/E= -go.opentelemetry.io/otel/trace v1.28.0 h1:GhQ9cUuQGmNDd5BTCP2dAvv75RdMxEfTmYejp+lkx9g= -go.opentelemetry.io/otel/trace v1.28.0/go.mod h1:jPyXzNPg6da9+38HEwElrQiHlVMTnVfM3/yv2OlIHaI= +go.opentelemetry.io/otel v1.29.0 h1:PdomN/Al4q/lN6iBJEN3AwPvUiHPMlt93c8bqTG5Llw= +go.opentelemetry.io/otel v1.29.0/go.mod h1:N/WtXPs1CNCUEx+Agz5uouwCba+i+bJGFicT8SR4NP8= +go.opentelemetry.io/otel/metric v1.29.0 h1:vPf/HFWTNkPu1aYeIsc98l4ktOQaL6LeSoeV2g+8YLc= +go.opentelemetry.io/otel/metric v1.29.0/go.mod h1:auu/QWieFVWx+DmQOUMgj0F8LHWdgalxXqvp7BII/W8= +go.opentelemetry.io/otel/sdk v1.29.0 h1:vkqKjk7gwhS8VaWb0POZKmIEDimRCMsopNYnriHyryo= +go.opentelemetry.io/otel/sdk v1.29.0/go.mod h1:pM8Dx5WKnvxLCb+8lG1PRNIDxu9g9b9g59Qr7hfAAok= +go.opentelemetry.io/otel/trace v1.29.0 h1:J/8ZNK4XgR7a21DZUAsbF8pZ5Jcw1VhACmnYt39JTi4= +go.opentelemetry.io/otel/trace v1.29.0/go.mod h1:eHl3w0sp3paPkYstJOmAimxhiFXPg+MMTlEh3nsQgWQ= +go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE= +go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI= +go.uber.org/multierr v1.9.0/go.mod h1:X2jQV1h+kxSjClGpnseKVIxpmcjrj7MNnI0bnlfKTVQ= golang.org/x/arch v0.18.0 h1:WN9poc33zL4AzGxqf8VtpKUnGvMi8O9lhNyBMF/85qc= golang.org/x/arch v0.18.0/go.mod h1:bdwinDaKcfZUGpH09BB7ZmOfhalA8lQdzl62l8gGWsk= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= diff --git a/ingestor/main.go b/ingestor/main.go index 5da5598..66e8449 100644 --- a/ingestor/main.go +++ b/ingestor/main.go @@ -12,16 +12,14 @@ import ( "log/slog" "github.com/IBM/sarama" + "github.com/MinuteHanD/log-pipeline/config" "github.com/gin-gonic/gin" ) const ( - KafkaTopic = "raw_logs" MaxMessageSize = 64 * 1024 // max message size 64 kb ) -var KafkaBrokers = os.Getenv("KAFKA_BROKERS") - type LogEntry struct { Timestamp string `json:"timestamp" binding:"required"` Level string `json:"level" binding:"required"` @@ -180,17 +178,20 @@ func (v *LogValidator) ValidateComplete(data []byte) ValidationResult { } func main() { - // Initialize a structured logger that outputs JSON to standard output. logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + cfg, err := config.LoadConfig() + if err != nil { + logger.Error("Failed to load configuration", "error", err) + os.Exit(1) + } + config := sarama.NewConfig() config.Producer.Return.Successes = true config.Producer.RequiredAcks = sarama.WaitForAll config.Version = sarama.V2_8_0_0 - KafkaBrokers := os.Getenv("KAFKA_BROKERS") - brokers := strings.Split(KafkaBrokers, ",") - producer, err := sarama.NewSyncProducer(brokers, config) + producer, err := sarama.NewSyncProducer(cfg.Kafka.Brokers, config) if err != nil { logger.Error("Failed to start Sarama producer", "error", err) @@ -201,10 +202,6 @@ func main() { validator := NewLogValidator() router := gin.Default() - // Replace Gin's default logger with our structured logger or disable it if not needed. - // For simplicity, we'll keep Gin's logger for request logging, but our app logs will be structured. - // router.Use(gin.Logger()) - router.POST("/log", func(c *gin.Context) { body, err := c.GetRawData() if err != nil { @@ -227,13 +224,13 @@ func main() { } msg := &sarama.ProducerMessage{ - Topic: KafkaTopic, + Topic: cfg.Kafka.Topics.Raw, Value: sarama.ByteEncoder(body), } partition, offset, err := producer.SendMessage(msg) if err != nil { - logger.Error("Failed to send message to Kafka", "error", err, "topic", KafkaTopic) + logger.Error("Failed to send message to Kafka", "error", err, "topic", cfg.Kafka.Topics.Raw) c.JSON(http.StatusInternalServerError, gin.H{ "error": "failed to send message to kafka", "details": err.Error(), @@ -241,7 +238,7 @@ func main() { return } - logger.Info("Log received and validated successfully", "topic", KafkaTopic, "partition", partition, "offset", offset) + logger.Info("Log received and validated successfully", "topic", cfg.Kafka.Topics.Raw, "partition", partition, "offset", offset) c.JSON(http.StatusOK, gin.H{ "message": "log recieved and validated successfully", "partition": partition, @@ -260,8 +257,12 @@ func main() { }) }) - logger.Info("Ingestor service starting", "port", 8081) - if err := router.Run(":8081"); err != nil { + router.GET("/health", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + logger.Info("Ingestor service starting", "port", cfg.Ingestor.HTTPPort) + if err := router.Run(fmt.Sprintf(":%d", cfg.Ingestor.HTTPPort)); err != nil { logger.Error("Failed to run Gin server", "error", err) os.Exit(1) } diff --git a/kafka/dlq.go b/kafka/dlq.go new file mode 100644 index 0000000..450d1b9 --- /dev/null +++ b/kafka/dlq.go @@ -0,0 +1,40 @@ +package kafka + +import ( + "fmt" + + "github.com/IBM/sarama" + "log/slog" +) + +func SendToDLQ(logger *slog.Logger, producer sarama.SyncProducer, dlqTopic string, originalMessage *sarama.ConsumerMessage, processingError error) { + dlqMessage := &sarama.ProducerMessage{ + Topic: dlqTopic, + Value: sarama.ByteEncoder(originalMessage.Value), + Headers: []sarama.RecordHeader{ + { + Key: []byte("error"), + Value: []byte(processingError.Error()), + }, + { + Key: []byte("original_topic"), + Value: []byte(originalMessage.Topic), + }, + { + Key: []byte("original_partition"), + Value: []byte(fmt.Sprintf("%d", originalMessage.Partition)), + }, + { + Key: []byte("original_offset"), + Value: []byte(fmt.Sprintf("%d", originalMessage.Offset)), + }, + }, + } + + _, _, err := producer.SendMessage(dlqMessage) + if err != nil { + logger.Error("CRITICAL: Failed to send message to DLQ", "topic", dlqTopic, "error", err) + } else { + logger.Info("Message sent to DLQ", "topic", dlqTopic, "reason", processingError.Error()) + } +} diff --git a/log-generator/Dockerfile b/log-generator/Dockerfile new file mode 100644 index 0000000..2a8d286 --- /dev/null +++ b/log-generator/Dockerfile @@ -0,0 +1,12 @@ +FROM golang:1.24.3-alpine + +WORKDIR /app + +COPY go.mod go.sum ./ +RUN go mod download + +COPY ./log-generator . + +RUN go build -o main . + +CMD ["./main"] diff --git a/log-generator/main.go b/log-generator/main.go new file mode 100644 index 0000000..f2d44db --- /dev/null +++ b/log-generator/main.go @@ -0,0 +1,125 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "log/slog" + "math/rand" + "net/http" + "os" + "strings" + "time" +) + +var ( + services = []string{"api-gateway", "user-service", "payment-processor", "auth-service", "notification-sender"} + levels = []string{"INFO", "WARN", "ERROR", "DEBUG"} +) + +type LogEntry struct { + Timestamp string `json:"timestamp"` + Level string `json:"level"` + Message string `json:"message"` + Service string `json:"service"` +} + +func main() { + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + logger.Info("Log generator service starting up...") + + ingestorURL := os.Getenv("INGESTOR_URL") + if ingestorURL == "" { + logger.Error("INGESTOR_URL environment variable is not set. Can't send logs.") + os.Exit(1) + } + ingestorHealthURL := strings.Replace(ingestorURL, "/log", "/health", 1) + + for { + resp, err := http.Get(ingestorHealthURL) + if err == nil && resp.StatusCode == http.StatusOK { + resp.Body.Close() + logger.Info("Ingestor service is ready.") + break + } + if resp != nil { + resp.Body.Close() + } + logger.Info("Waiting for ingestor service to be ready...") + time.Sleep(2 * time.Second) + } + + logger.Info("Logs will be sent to", "url", ingestorURL) + + source := rand.NewSource(time.Now().UnixNano()) + random := rand.New(source) + + for { + + sleepDuration := time.Duration(500+random.Intn(2500)) * time.Millisecond + time.Sleep(sleepDuration) + + logEntry := generateRandomLog(random) + + payload, err := json.Marshal(logEntry) + if err != nil { + + logger.Error("Failed to marshal log entry to JSON", "error", err) + continue + } + + resp, err := http.Post(ingestorURL, "application/json", bytes.NewReader(payload)) + if err != nil { + logger.Error("Failed to send log to ingestor", "error", err, "service", logEntry.Service) + continue + } + + if resp.StatusCode != http.StatusOK { + logger.Warn("Ingestor returned a non-OK status", "status_code", resp.StatusCode, "service", logEntry.Service) + } else { + logger.Info("Successfully sent a log", "service", logEntry.Service, "level", logEntry.Level) + } + resp.Body.Close() + } +} + +func generateRandomLog(random *rand.Rand) LogEntry { + service := services[random.Intn(len(services))] + level := levels[random.Intn(len(levels))] + message := generateRandomMessage(service, level, random) + + return LogEntry{ + Timestamp: time.Now().UTC().Format(time.RFC3339), + Level: level, + Message: message, + Service: service, + } +} + +func generateRandomMessage(service, level string, random *rand.Rand) string { + userID := 1000 + random.Intn(9000) + switch level { + case "ERROR": + errorMessages := []string{ + "Failed to connect to database: timeout expired", + "Payment failed for user %d: insufficient funds", + "Authentication token expired for user %d", + "Null pointer exception at process step %d", + } + return fmt.Sprintf(errorMessages[random.Intn(len(errorMessages))], userID) + case "WARN": + warnMessages := []string{ + "High latency detected on upstream service: %dms", + "Disk space running low: %d%% remaining", + "API rate limit approaching for client %d", + } + return fmt.Sprintf(warnMessages[random.Intn(len(warnMessages))], 50+random.Intn(200)) + default: + infoMessages := []string{ + "User %d logged in successfully.", + "Processed new order #%d for user %d.", + "Retrieved user profile for user %d.", + } + return fmt.Sprintf(infoMessages[random.Intn(len(infoMessages))], userID, 100+random.Intn(500), userID) + } +} diff --git a/log-generator/main_test.go b/log-generator/main_test.go new file mode 100644 index 0000000..09b7e59 --- /dev/null +++ b/log-generator/main_test.go @@ -0,0 +1,9 @@ +package main + +import ( + "testing" +) + +func TestGenerateRandomLog(t *testing.T) { + +} diff --git a/main_integration_test.go b/main_integration_test.go index 758f511..7dee5c8 100644 --- a/main_integration_test.go +++ b/main_integration_test.go @@ -34,7 +34,7 @@ func TestPipeline_Integration(t *testing.T) { ingestorReady := false for i := 0; i < 30; i++ { - resp, err := http.Get("http://localhost:8081/validation-rules") + resp, err := http.Get("http://localhost:8081/health") if err == nil && resp.StatusCode == http.StatusOK { ingestorReady = true resp.Body.Close() diff --git a/parser/main.go b/parser/main.go index b5489fe..6c0e962 100644 --- a/parser/main.go +++ b/parser/main.go @@ -14,12 +14,11 @@ import ( "log/slog" "github.com/IBM/sarama" + "github.com/MinuteHanD/log-pipeline/config" + "github.com/MinuteHanD/log-pipeline/kafka" ) const ( - InputTopic = "raw_logs" - OutputTopic = "parsed_logs" - DLQTopic = "raw_logs_dlq" ConsumerGroup = "parser-group" ) @@ -49,9 +48,9 @@ type LogProcessor struct { logger *slog.Logger } -func NewLogProcessor(logger *slog.Logger) *LogProcessor { +func NewLogProcessor(logger *slog.Logger, version string) *LogProcessor { return &LogProcessor{ - version: "1.0.0", + version: version, logger: logger, } } @@ -157,64 +156,31 @@ func generateMessageHash(message string) string { return string(rune(hash)) } -func sendToDLQ(logger *slog.Logger, producer sarama.SyncProducer, originalMessage *sarama.ConsumerMessage, processingError error) { - dlqMessage := &sarama.ProducerMessage{ - Topic: DLQTopic, - Value: sarama.ByteEncoder(originalMessage.Value), - Headers: []sarama.RecordHeader{ - { - Key: []byte("error"), - Value: []byte(processingError.Error()), - }, - { - Key: []byte("original_topic"), - Value: []byte(originalMessage.Topic), - }, - { - Key: []byte("original_partition"), - Value: []byte(fmt.Sprintf("%d", originalMessage.Partition)), - }, - { - Key: []byte("original_offset"), - Value: []byte(fmt.Sprintf("%d", originalMessage.Offset)), - }, - }, - } - _, _, err := producer.SendMessage(dlqMessage) - if err != nil { - logger.Error("CRITICAL: Failed to send message to DLQ", "topic", DLQTopic, "error", err) - } else { - logger.Info("Message sent to DLQ", "topic", DLQTopic, "reason", processingError.Error()) - } -} func main() { logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) - logger.Info("Starting parser service...") - - kafkaBrokers := os.Getenv("KAFKA_BROKERS") - if kafkaBrokers == "" { - logger.Error("KAFKA_BROKERS environment variable not set") + cfg, err := config.LoadConfig() + if err != nil { + logger.Error("Failed to load configuration", "error", err) os.Exit(1) } - brokers := strings.Split(kafkaBrokers, ",") config := sarama.NewConfig() config.Version = sarama.V2_8_0_0 config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initial = sarama.OffsetOldest - consumerGroup, err := sarama.NewConsumerGroup(brokers, ConsumerGroup, config) + consumerGroup, err := sarama.NewConsumerGroup(cfg.Kafka.Brokers, ConsumerGroup, config) if err != nil { logger.Error("Failed to create consumer group", "error", err) os.Exit(1) } defer consumerGroup.Close() - producer, err := newProducer(brokers) + producer, err := newProducer(cfg.Kafka.Brokers) if err != nil { logger.Error("Failed to create producer", "error", err) os.Exit(1) @@ -223,13 +189,14 @@ func main() { ctx, cancel := context.WithCancel(context.Background()) - processor := NewLogProcessor(logger) + processor := NewLogProcessor(logger, cfg.Parser.Version) handler := &logHandler{ producer: producer, processor: processor, ready: make(chan bool), logger: logger, + cfg: cfg, } wg := &sync.WaitGroup{} @@ -238,7 +205,7 @@ func main() { go func() { defer wg.Done() for { - if err := consumerGroup.Consume(ctx, []string{InputTopic}, handler); err != nil { + if err := consumerGroup.Consume(ctx, []string{handler.cfg.Kafka.Topics.Raw}, handler); err != nil { logger.Error("Error from consumer", "error", err) } if ctx.Err() != nil { @@ -273,6 +240,7 @@ type logHandler struct { processor *LogProcessor ready chan bool logger *slog.Logger + cfg *config.Config } func (h *logHandler) Setup(_ sarama.ConsumerGroupSession) error { @@ -292,7 +260,7 @@ func (h *logHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sar if err := json.Unmarshal(message.Value, &incomingLog); err != nil { h.logger.Error("Failed to unmarshal incoming log", "error", err, "topic", message.Topic, "offset", message.Offset) - sendToDLQ(h.logger, h.producer, message, err) + kafka.SendToDLQ(h.logger, h.producer, h.cfg.Kafka.Topics.RawDLQ, message, err) session.MarkMessage(message, "") continue @@ -302,7 +270,7 @@ func (h *logHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sar if err != nil { h.logger.Error("Failed to process log", "error", err, "topic", message.Topic, "offset", message.Offset) - sendToDLQ(h.logger, h.producer, message, err) + kafka.SendToDLQ(h.logger, h.producer, h.cfg.Kafka.Topics.RawDLQ, message, err) session.MarkMessage(message, "") continue @@ -316,15 +284,15 @@ func (h *logHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sar } producerMsg := &sarama.ProducerMessage{ - Topic: OutputTopic, + Topic: h.cfg.Kafka.Topics.Parsed, Value: sarama.ByteEncoder(enrichedLogBytes), } partition, offset, err := h.producer.SendMessage(producerMsg) if err != nil { - h.logger.Error("Failed to send processed log to Kafka", "error", err, "topic", OutputTopic, "log_id", parsedLog.ID) + h.logger.Error("Failed to send processed log to Kafka", "error", err, "topic", h.cfg.Kafka.Topics.Parsed, "log_id", parsedLog.ID) } else { - h.logger.Info("Successfully sent processed log to Kafka", "topic", OutputTopic, "partition", partition, "offset", offset, "log_id", parsedLog.ID) + h.logger.Info("Successfully sent processed log to Kafka", "topic", h.cfg.Kafka.Topics.Parsed, "partition", partition, "offset", offset, "log_id", parsedLog.ID) } session.MarkMessage(message, "") diff --git a/parser/main_test.go b/parser/main_test.go index dc458f1..d338fe4 100644 --- a/parser/main_test.go +++ b/parser/main_test.go @@ -13,7 +13,7 @@ func newDiscardLogger() *slog.Logger { } func TestLogProcessor_normalizeTimestamp(t *testing.T) { - processor := NewLogProcessor(newDiscardLogger()) + processor := NewLogProcessor(newDiscardLogger(), "test-version") testCases := []struct { name string @@ -69,7 +69,7 @@ func TestLogProcessor_normalizeTimestamp(t *testing.T) { } func TestLogProcessor_normalizeLogLevel(t *testing.T) { - processor := NewLogProcessor(newDiscardLogger()) + processor := NewLogProcessor(newDiscardLogger(), "test-version") testCases := []struct { name string @@ -94,7 +94,7 @@ func TestLogProcessor_normalizeLogLevel(t *testing.T) { } func TestLogProcessor_ProcessLog(t *testing.T) { - processor := NewLogProcessor(newDiscardLogger()) + processor := NewLogProcessor(newDiscardLogger(), "test-version") rawLog := IncomingLogEntry{ Timestamp: "2025-07-01T12:00:00Z", diff --git a/storage-writer/main.go b/storage-writer/main.go index 73006e8..f736cf8 100644 --- a/storage-writer/main.go +++ b/storage-writer/main.go @@ -17,13 +17,14 @@ import ( "github.com/IBM/sarama" "github.com/elastic/go-elasticsearch/v8" "github.com/elastic/go-elasticsearch/v8/esapi" + "github.com/MinuteHanD/log-pipeline/config" + "github.com/MinuteHanD/log-pipeline/kafka" ) + + const ( - InputTopic = "parsed_logs" - ConsumerGroup = "storage-writer-group" - ElasticsearchDLQTopic = "parsed_logs_dlq" // Dead-Letter Queue topic for Elasticsearch failures - IndexPrefix = "logs" + ConsumerGroup = "storage-writer-group" ) type ParsedLog struct { @@ -47,14 +48,16 @@ type ElasticsearchDocument struct { } type IndexManager struct { - client *elasticsearch.Client - logger *slog.Logger + client *elasticsearch.Client + logger *slog.Logger + indexPrefix string } -func NewIndexManager(client *elasticsearch.Client, logger *slog.Logger) *IndexManager { +func NewIndexManager(client *elasticsearch.Client, logger *slog.Logger, indexPrefix string) *IndexManager { return &IndexManager{ - client: client, - logger: logger, + client: client, + logger: logger, + indexPrefix: indexPrefix, } } @@ -125,7 +128,7 @@ func (im *IndexManager) GetIndexname(timestamp time.Time) string { im.logger.Warn("Invalid timestamp (zero value), using current time for index name") t = time.Now() } - return fmt.Sprintf("%s-%s", IndexPrefix, t.Format("2006.01.02")) + return fmt.Sprintf("%s-%s", im.indexPrefix, t.Format("2006.01.02")) } type LogStorageProcessor struct { @@ -135,11 +138,11 @@ type LogStorageProcessor struct { logger *slog.Logger } -func NewLogStorageProcessor(client *elasticsearch.Client, logger *slog.Logger) *LogStorageProcessor { +func NewLogStorageProcessor(client *elasticsearch.Client, logger *slog.Logger, version string, indexPrefix string) *LogStorageProcessor { return &LogStorageProcessor{ esClient: client, - indexManager: NewIndexManager(client, logger), - version: "1.0.0", + indexManager: NewIndexManager(client, logger, indexPrefix), + version: version, logger: logger, } } @@ -192,17 +195,14 @@ func (lsp *LogStorageProcessor) ProcessAndStore(messageValue []byte) error { func main() { logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) - logger.Info("Starting storage writer...") - - kafkaEnv := os.Getenv("KAFKA_BROKERS") - esEnv := os.Getenv("ELASTICSEARCH_URL") - if kafkaEnv == "" || esEnv == "" { - logger.Error("Environment variables KAFKA_BROKERS and ELASTICSEARCH_URL must be set") + cfg, err := config.LoadConfig() + if err != nil { + logger.Error("Failed to load configuration", "error", err) os.Exit(1) } esClient, err := elasticsearch.NewClient(elasticsearch.Config{ - Addresses: []string{esEnv}, + Addresses: []string{cfg.Elasticsearch.URL}, }) if err != nil { logger.Error("Error creating Elasticsearch client", "error", err) @@ -226,11 +226,9 @@ func main() { config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initial = sarama.OffsetOldest - brokers := strings.Split(kafkaEnv, ",") - var consumerGroup sarama.ConsumerGroup for i := 0; i < 10; i++ { - consumerGroup, err = sarama.NewConsumerGroup(brokers, ConsumerGroup, config) + consumerGroup, err = sarama.NewConsumerGroup(cfg.Kafka.Brokers, ConsumerGroup, config) if err == nil { break } @@ -243,7 +241,7 @@ func main() { } defer consumerGroup.Close() - producer, err := newProducer(brokers) + producer, err := newProducer(cfg.Kafka.Brokers) if err != nil { logger.Error("Failed to create producer for DLQ", "error", err) os.Exit(1) @@ -253,13 +251,14 @@ func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - processor := NewLogStorageProcessor(esClient, logger) + processor := NewLogStorageProcessor(esClient, logger, cfg.StorageWriter.Version, cfg.StorageWriter.IndexPrefix) handler := &logStorageHandler{ processor: processor, producer: producer, ready: make(chan bool), logger: logger, + cfg: cfg, } wg := &sync.WaitGroup{} @@ -267,7 +266,7 @@ func main() { go func() { defer wg.Done() for { - if err := consumerGroup.Consume(ctx, []string{InputTopic}, handler); err != nil { + if err := consumerGroup.Consume(ctx, []string{cfg.Kafka.Topics.Parsed}, handler); err != nil { logger.Error("Error from consumer", "error", err) } if ctx.Err() != nil { @@ -297,43 +296,14 @@ func newProducer(brokers []string) (sarama.SyncProducer, error) { return sarama.NewSyncProducer(brokers, config) } -func sendToDLQ(logger *slog.Logger, producer sarama.SyncProducer, originalMessage *sarama.ConsumerMessage, processingError error) { - dlqMessage := &sarama.ProducerMessage{ - Topic: ElasticsearchDLQTopic, - Value: sarama.ByteEncoder(originalMessage.Value), - Headers: []sarama.RecordHeader{ - { - Key: []byte("error"), - Value: []byte(processingError.Error()), - }, - { - Key: []byte("original_topic"), - Value: []byte(originalMessage.Topic), - }, - { - Key: []byte("original_partition"), - Value: []byte(fmt.Sprintf("%d", originalMessage.Partition)), - }, - { - Key: []byte("original_offset"), - Value: []byte(fmt.Sprintf("%d", originalMessage.Offset)), - }, - }, - } - _, _, err := producer.SendMessage(dlqMessage) - if err != nil { - logger.Error("CRITICAL: Failed to send message to Elasticsearch DLQ", "topic", ElasticsearchDLQTopic, "error", err) - } else { - logger.Info("Message sent to Elasticsearch DLQ", "topic", ElasticsearchDLQTopic, "reason", processingError.Error()) - } -} type logStorageHandler struct { processor *LogStorageProcessor producer sarama.SyncProducer ready chan bool logger *slog.Logger + cfg *config.Config } func (handler *logStorageHandler) Setup(_ sarama.ConsumerGroupSession) error { @@ -353,7 +323,7 @@ func (handler *logStorageHandler) ConsumeClaim(session sarama.ConsumerGroupSessi handler.logger.Error("Failed to process and store log", "error", err, "topic", message.Topic, "offset", message.Offset) // Send the failed message to the Dead-Letter Queue - sendToDLQ(handler.logger, handler.producer, message, err) + kafka.SendToDLQ(handler.logger, handler.producer, handler.cfg.Kafka.Topics.ParsedDLQ, message, err) session.MarkMessage(message, "") continue diff --git a/storage-writer/main_test.go b/storage-writer/main_test.go index 374eb81..11e0dbd 100644 --- a/storage-writer/main_test.go +++ b/storage-writer/main_test.go @@ -13,7 +13,7 @@ func newDiscardLogger() *slog.Logger { func TestIndexManager_GetIndexname(t *testing.T) { - manager := NewIndexManager(nil, newDiscardLogger()) + manager := NewIndexManager(nil, newDiscardLogger(), "logs") testCases := []struct { name string