11package main
22
33import (
4+ "context"
5+ "errors"
46 "fmt"
7+ "log/slog"
58 "os"
9+ "os/signal"
10+ "time"
611
712 "github.com/spf13/pflag"
13+ "golang.org/x/sys/unix"
14+
815 "github.com/scality/log-courier/pkg/logcourier"
16+ "github.com/scality/log-courier/pkg/util"
917)
1018
1119func main () {
20+ os .Exit (run ())
21+ }
22+
23+ // buildProcessorConfig creates processor config from ConfigSpec
24+ func buildProcessorConfig (logger * slog.Logger ) logcourier.Config {
25+ return logcourier.Config {
26+ ClickHouseHosts : logcourier .ConfigSpec .GetStringSlice ("clickhouse.url" ),
27+ ClickHouseUsername : logcourier .ConfigSpec .GetString ("clickhouse.username" ),
28+ ClickHousePassword : logcourier .ConfigSpec .GetString ("clickhouse.password" ),
29+ ClickHouseTimeout : time .Duration (logcourier .ConfigSpec .GetInt ("clickhouse.timeout-seconds" )) * time .Second ,
30+ CountThreshold : logcourier .ConfigSpec .GetInt ("consumer.count-threshold" ),
31+ TimeThresholdSec : logcourier .ConfigSpec .GetInt ("consumer.time-threshold-seconds" ),
32+ DiscoveryInterval : time .Duration (logcourier .ConfigSpec .GetInt ("consumer.discovery-interval-seconds" )) * time .Second ,
33+ DiscoveryIntervalJitterFactor : logcourier .ConfigSpec .GetFloat64 ("consumer.discovery-interval-jitter-factor" ),
34+ NumWorkers : logcourier .ConfigSpec .GetInt ("consumer.num-workers" ),
35+ MaxRetries : logcourier .ConfigSpec .GetInt ("retry.max-retries" ),
36+ InitialBackoff : time .Duration (logcourier .ConfigSpec .GetInt ("retry.initial-backoff-seconds" )) * time .Second ,
37+ MaxBackoff : time .Duration (logcourier .ConfigSpec .GetInt ("retry.max-backoff-seconds" )) * time .Second ,
38+ BackoffJitterFactor : logcourier .ConfigSpec .GetFloat64 ("retry.backoff-jitter-factor" ),
39+ S3Endpoint : logcourier .ConfigSpec .GetString ("s3.endpoint" ),
40+ S3AccessKeyID : logcourier .ConfigSpec .GetString ("s3.access-key-id" ),
41+ S3SecretAccessKey : logcourier .ConfigSpec .GetString ("s3.secret-access-key" ),
42+ Logger : logger ,
43+ }
44+ }
45+
46+ func run () int {
47+ // Add command-line flags
1248 logcourier .ConfigSpec .AddFlag (pflag .CommandLine , "log-level" , "log-level" )
1349
1450 configFileFlag := pflag .String ("config-file" , "" , "Path to configuration file" )
1551 pflag .Parse ()
1652
53+ // Load configuration
1754 configFile := * configFileFlag
1855 if configFile == "" {
1956 configFile = os .Getenv ("LOG_COURIER_CONFIG_FILE" )
@@ -23,16 +60,81 @@ func main() {
2360 if err != nil {
2461 fmt .Fprintf (os .Stderr , "Configuration error: %v\n " , err )
2562 pflag .Usage ()
26- os . Exit ( 2 )
63+ return 2
2764 }
2865
66+ // Validate configuration
2967 err = logcourier .ValidateConfig ()
3068 if err != nil {
3169 fmt .Fprintf (os .Stderr , "Configuration validation error: %v\n " , err )
32- pflag .Usage ()
33- os .Exit (2 )
70+ return 2
71+ }
72+
73+ // Set up logger
74+ logLevel := util .ParseLogLevel (logcourier .ConfigSpec .GetString ("log-level" ))
75+ logger := slog .New (slog .NewJSONHandler (os .Stdout , & slog.HandlerOptions {
76+ Level : logLevel ,
77+ }))
78+
79+ // Get shutdown timeout from config
80+ shutdownTimeout := time .Duration (logcourier .ConfigSpec .GetInt ("shutdown-timeout-seconds" )) * time .Second
81+
82+ // Create processor
83+ ctx := context .Background ()
84+ processorCfg := buildProcessorConfig (logger )
85+
86+ processor , err := logcourier .NewProcessor (ctx , processorCfg )
87+ if err != nil {
88+ logger .Error ("failed to create processor" , "error" , err )
89+ return 1
90+ }
91+ defer func () {
92+ if err := processor .Close (); err != nil {
93+ logger .Error ("failed to close processor" , "error" , err )
94+ }
95+ }()
96+
97+ // Set up signal handling for graceful shutdown
98+ ctx , cancel := context .WithCancel (ctx )
99+ defer cancel ()
100+
101+ signalsChan := make (chan os.Signal , 1 )
102+ signal .Notify (signalsChan , unix .SIGINT , unix .SIGTERM )
103+
104+ // Start processor in goroutine
105+ errChan := make (chan error )
106+ go func () {
107+ errChan <- processor .Run (ctx )
108+ }()
109+
110+ // Wait for signal or error
111+ select {
112+ case sig := <- signalsChan :
113+ logger .Info ("signal received" , "signal" , sig )
114+ cancel ()
115+
116+ // Wait for processor to stop gracefully (with timeout)
117+ shutdownTimer := time .NewTimer (shutdownTimeout )
118+ defer shutdownTimer .Stop ()
119+
120+ select {
121+ case <- shutdownTimer .C :
122+ logger .Warn ("shutdown timeout exceeded, forcing exit" )
123+ return 1
124+ case err := <- errChan :
125+ if err != nil && ! errors .Is (err , context .Canceled ) {
126+ logger .Error ("processor stopped with error" , "error" , err )
127+ return 1
128+ }
129+ }
130+
131+ case err := <- errChan :
132+ if err != nil && ! errors .Is (err , context .Canceled ) {
133+ logger .Error ("processor error" , "error" , err )
134+ return 1
135+ }
34136 }
35137
36- // TODO: Initialize and run consumer
37- fmt . Println ( "log-courier started (stub)" )
138+ logger . Info ( "log-courier stopped" )
139+ return 0
38140}
0 commit comments