From 4d606179ef76837decdceef0def8ac20f2d7b10d Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Tue, 7 Mar 2023 12:29:25 +0300 Subject: [PATCH] More fixes --- pkg/binary/output_binary.go | 4 +++- pkg/binary/tcp_client.go | 28 +++++++++++++--------------- pkg/elasticsearch/elasticsearch.go | 10 ++++++---- pkg/emitter/emitter.go | 16 +++++++++------- pkg/file/input_file.go | 18 ++++++++++-------- pkg/file/output_file.go | 7 ++++--- pkg/file/output_s3.go | 10 ++++++---- pkg/file/s3_reader.go | 12 +++++------- pkg/http/input_http.go | 9 ++++++--- pkg/http/output_http.go | 10 ++++++---- pkg/pro/pro.go | 1 + 11 files changed, 69 insertions(+), 56 deletions(-) diff --git a/pkg/binary/output_binary.go b/pkg/binary/output_binary.go index 43c97ba..9effcb2 100644 --- a/pkg/binary/output_binary.go +++ b/pkg/binary/output_binary.go @@ -15,6 +15,8 @@ const ( initialDynamicWorkers = 10 ) +var outputLogger = log.With().Str("component", "output_binary").Logger() + // BinaryOutputConfig struct for holding binary output configuration type BinaryOutputConfig struct { Workers int `json:"output-binary-workers"` @@ -162,7 +164,7 @@ func (o *BinaryOutput) sendRequest(client *TCPClient, msg *plugin.Message) { stop := time.Now() if err != nil { - log.Error().Err(err).Msg("Request error") + outputLogger.Error().Err(err).Msg("Request error") } if o.config.TrackResponses { diff --git a/pkg/binary/tcp_client.go b/pkg/binary/tcp_client.go index 12fd342..adec39e 100644 --- a/pkg/binary/tcp_client.go +++ b/pkg/binary/tcp_client.go @@ -6,8 +6,6 @@ import ( "net" "syscall" "time" - - "github.com/rs/zerolog/log" ) const ( @@ -77,7 +75,7 @@ func (c *TCPClient) Disconnect() { c.conn.Close() c.conn = nil - log.Warn().Msgf("Disconnected: %s", c.baseURL) + outputLogger.Warn().Msgf("Disconnected: %s", c.baseURL) } } @@ -91,10 +89,10 @@ func (c *TCPClient) isAlive() bool { if err == nil { return true } else if err == io.EOF { - log.Warn().Msg("connection closed, reconnecting") + outputLogger.Warn().Msg("connection closed, reconnecting") return false } else if err == syscall.EPIPE { - log.Warn().Msg("broken pipe, reconnecting") + outputLogger.Warn().Msg("broken pipe, reconnecting") return false } @@ -106,18 +104,18 @@ func (c *TCPClient) Send(data []byte) (response []byte, err error) { // Don't exit on panic defer func() { if r := recover(); r != nil { - log.Error().Msgf("PANIC: pkg: %v", r) + outputLogger.Error().Msgf("PANIC: pkg: %v", r) if _, ok := r.(error); !ok { - log.Error().Stack().Msgf("faile to send request: %s", string(data)) + outputLogger.Error().Stack().Msgf("faile to send request: %s", string(data)) } } }() if c.conn == nil || !c.isAlive() { - log.Info().Msgf("Connecting: %s", c.baseURL) + outputLogger.Info().Msgf("Connecting: %s", c.baseURL) if err = c.Connect(); err != nil { - log.Error().Err(err).Msgf("Connection error: %s", c.baseURL) + outputLogger.Error().Err(err).Msgf("Connection error: %s", c.baseURL) return } } @@ -127,11 +125,11 @@ func (c *TCPClient) Send(data []byte) (response []byte, err error) { c.conn.SetWriteDeadline(timeout) if c.config.Debug { - log.Debug().Msgf("Sending: %s", string(data)) + outputLogger.Debug().Msgf("Sending: %s", string(data)) } if _, err = c.conn.Write(data); err != nil { - log.Error().Err(err).Msgf("Write error: %s", c.baseURL) + outputLogger.Error().Err(err).Msgf("Write error: %s", c.baseURL) return } @@ -162,7 +160,7 @@ func (c *TCPClient) Send(data []byte) (response []byte, err error) { if err == io.EOF { break } else if err != nil { - log.Error().Err(err).Msgf("Read error: %s", c.baseURL) + outputLogger.Error().Err(err).Msgf("Read error: %s", c.baseURL) break } @@ -170,7 +168,7 @@ func (c *TCPClient) Send(data []byte) (response []byte, err error) { } if readBytes >= maxResponseSize { - log.Error().Msgf("Body is more than the max size: %d", maxResponseSize) + outputLogger.Error().Msgf("Body is more than the max size: %d", maxResponseSize) break } @@ -179,7 +177,7 @@ func (c *TCPClient) Send(data []byte) (response []byte, err error) { } if err != nil { - log.Error().Err(err).Msgf("Response read error") + outputLogger.Error().Err(err).Msgf("Response read error") return } @@ -191,7 +189,7 @@ func (c *TCPClient) Send(data []byte) (response []byte, err error) { copy(payload, c.respBuf[:readBytes]) if c.config.Debug { - log.Debug().Msgf("Received: %s", string(payload)) + outputLogger.Debug().Msgf("Received: %s", string(payload)) } return payload, err diff --git a/pkg/elasticsearch/elasticsearch.go b/pkg/elasticsearch/elasticsearch.go index 09ee1cb..e578d7b 100644 --- a/pkg/elasticsearch/elasticsearch.go +++ b/pkg/elasticsearch/elasticsearch.go @@ -13,6 +13,8 @@ import ( "github.com/rs/zerolog/log" ) +var logger = log.With().Str("component", "elasticsearch").Logger() + type ESUriErorr struct{} func (e *ESUriErorr) Error() string { @@ -89,7 +91,7 @@ func (p *ESPlugin) Init(URI string) { err, p.Index = parseURI(URI) if err != nil { - log.Fatal().Err(err).Msg("[ES] Can't initialize ElasticSearch plugin.") + logger.Fatal().Err(err).Msg("Can't initialize ElasticSearch plugin.") } p.eConn = elastigo.NewConn() @@ -102,7 +104,7 @@ func (p *ESPlugin) Init(URI string) { go p.ErrorHandler() - log.Info().Msg("[ES] Initialized Elasticsearch Plugin") + logger.Info().Msg("Initialized Elasticsearch Plugin") return } @@ -114,7 +116,7 @@ func (p *ESPlugin) IndexerShutdown() { func (p *ESPlugin) ErrorHandler() { for { errBuf := <-p.indexor.ErrorChannel - log.Error().Err(errBuf.Err).Msg("[ES] Error indexing document") + logger.Error().Err(errBuf.Err).Msg("Error indexing document") } } @@ -161,7 +163,7 @@ func (p *ESPlugin) ResponseAnalyze(req, resp []byte, start, stop time.Time) { j, err := json.Marshal(&esResp) if err != nil { - log.Error().Err(err).Msg("[ES] Error marshaling ESRequestResponse") + logger.Error().Err(err).Msg("Error marshaling ESRequestResponse") } else { p.indexor.Index(p.Index, "RequestResponse", "", "", "", &t, j) } diff --git a/pkg/emitter/emitter.go b/pkg/emitter/emitter.go index fef253b..24f4de7 100644 --- a/pkg/emitter/emitter.go +++ b/pkg/emitter/emitter.go @@ -19,6 +19,8 @@ import ( "github.com/rs/zerolog/log" ) +var logger = log.With().Str("component", "emitter").Logger() + // Emitter represents an abject to manage plugins communication type Emitter struct { sync.WaitGroup @@ -68,7 +70,7 @@ func (e *Emitter) Start(plugins *plugin.InOutPlugins) { go func() { defer e.Done() if err := e.CopyMulty(middleware, plugins.Outputs...); err != nil { - log.Error().Err(err).Msg("error during copy") + logger.Error().Err(err).Msg("error during copy") } }() } else { @@ -77,7 +79,7 @@ func (e *Emitter) Start(plugins *plugin.InOutPlugins) { go func(in plugin.Reader) { defer e.Done() if err := e.CopyMulty(in, plugins.Outputs...); err != nil { - log.Error().Err(err).Msg("error during copy") + logger.Error().Err(err).Msg("error during copy") } }(in) } @@ -118,16 +120,16 @@ func (e *Emitter) CopyMulty(src plugin.Reader, writers ...plugin.Writer) error { } meta := proto.PayloadMeta(msg.Meta) if len(meta) < 3 { - log.Warn().Msgf("[EMITTER] Found malformed record %q from %q", msg.Meta, src) + logger.Warn().Msgf("Found malformed record %q from %q", msg.Meta, src) continue } requestID := meta[1] // start a subroutine only when necessary if log.Logger.GetLevel() == zerolog.DebugLevel { - log.Debug().Msgf("[EMITTER] input: %s from: %s", byteutils.SliceToString(msg.Meta[:len(msg.Meta)-1]), src) + logger.Debug().Msgf("input: %s from: %s", byteutils.SliceToString(msg.Meta[:len(msg.Meta)-1]), src) } if modifier != nil { - log.Debug().Msgf("[EMITTER] modifier: %s from: %s", requestID, src) + logger.Debug().Msgf("modifier: %s from: %s", requestID, src) if proto.IsRequestPayload(msg.Meta) { msg.Data = modifier.Rewrite(msg.Data) // If modifier tells to skip request @@ -135,7 +137,7 @@ func (e *Emitter) CopyMulty(src plugin.Reader, writers ...plugin.Writer) error { filteredRequests.Set(requestID, []byte{}, 60) // continue } - log.Debug().Msgf("[EMITTER] Rewritten input: %s from: %s", requestID, src) + logger.Debug().Msgf("Rewritten input: %s from: %s", requestID, src) } else { _, err := filteredRequests.Get(requestID) if err == nil { @@ -155,7 +157,7 @@ func (e *Emitter) CopyMulty(src plugin.Reader, writers ...plugin.Writer) error { if e.config.SplitOutput { if e.config.RecognizeTCPSessions { if !pro.PRO { - log.Fatal().Msg("Detailed TCP sessions work only with PRO license") + logger.Fatal().Msg("Detailed TCP sessions work only with PRO license") } hasher := fnv.New32a() hasher.Write(meta[1]) diff --git a/pkg/file/input_file.go b/pkg/file/input_file.go index 9d1591d..735224c 100644 --- a/pkg/file/input_file.go +++ b/pkg/file/input_file.go @@ -27,6 +27,8 @@ import ( "github.com/rs/zerolog/log" ) +var inputLogger = log.With().Str("component", "input_file").Logger() + // InputFileConfig contains config of input file type InputFileConfig struct { InputFileLoop bool `json:"input-file-loop"` @@ -92,7 +94,7 @@ func (f *fileInputReader) parse(init chan struct{}) error { if err != nil { if err != io.EOF { - log.Logger.Error().Err(err).Msg("Error reading file") + inputLogger.Error().Err(err).Msg("Error reading file") } f.Close() @@ -110,7 +112,7 @@ func (f *fileInputReader) parse(init chan struct{}) error { meta := proto.PayloadMeta(asBytes) if len(meta) < 3 { - log.Warn().Msgf("Found malformed record, file: %s, line %d", f.path, lineNum) + inputLogger.Warn().Msgf("Found malformed record, file: %s, line %d", f.path, lineNum) buffer = bytes.Buffer{} continue } @@ -187,7 +189,7 @@ func newFileInputReader(path string, readDepth int, dryRun bool) *fileInputReade } if err != nil { - log.Error().Err(err).Msg("Error opening file") + inputLogger.Error().Err(err).Msg("Error opening file") return nil } @@ -195,7 +197,7 @@ func newFileInputReader(path string, readDepth int, dryRun bool) *fileInputReade if strings.HasSuffix(path, ".gz") { gzReader, err := gzip.NewReader(file) if err != nil { - log.Error().Err(err).Msg("Error opening compressed file") + inputLogger.Error().Err(err).Msg("Error opening compressed file") return nil } r.reader = bufio.NewReader(gzReader) @@ -269,7 +271,7 @@ func (i *FileInput) init() (err error) { resp, err := svc.ListObjects(params) if err != nil { - log.Error().Err(err).Msgf("Error while retrieving list of files from S3: %s", i.path) + inputLogger.Error().Err(err).Msgf("Error while retrieving list of files from S3: %s", i.path) return err } @@ -277,12 +279,12 @@ func (i *FileInput) init() (err error) { matches = append(matches, "s3://"+bucket+"/"+(*c.Key)) } } else if matches, err = filepath.Glob(i.path); err != nil { - log.Error().Err(err).Msgf("Error while retrieving list of files: %s", i.path) + inputLogger.Error().Err(err).Msgf("Error while retrieving list of files: %s", i.path) return } if len(matches) == 0 { - log.Error().Msgf("No files match pattern: %s", i.path) + inputLogger.Error().Msgf("No files match pattern: %s", i.path) return errors.New("no matching files") } @@ -432,7 +434,7 @@ func (i *FileInput) emit() { i.stats.Set("max_wait", time.Duration(maxWait)) i.stats.Set("min_wait", time.Duration(minWait)) - log.Info().Msgf("FileInput: end of file '%s'", i.path) + inputLogger.Info().Msgf("FileInput: end of file '%s'", i.path) if i.dryRun { fmt.Printf("Records found: %v\nFiles processed: %v\nBytes processed: %v\nMax wait: %v\nMin wait: %v\nFirst wait: %v\nIt will take `%v` to replay at current speed.\nFound %v records with out of order timestamp\n", diff --git a/pkg/file/output_file.go b/pkg/file/output_file.go index 47cdc12..3c433da 100644 --- a/pkg/file/output_file.go +++ b/pkg/file/output_file.go @@ -22,6 +22,7 @@ import ( "github.com/rs/zerolog/log" ) +var outputLogger = log.With().Str("component", "output_file").Logger() var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ") var instanceID string @@ -240,7 +241,7 @@ func (o *FileOutput) PluginWrite(msg *plugin.Message) (n int, err error) { } if err != nil { - log.Fatal().Err(err).Str("file", o.currentName).Msg("Cannot open file") + outputLogger.Fatal().Err(err).Str("file", o.currentName).Msg("Cannot open file") } o.QueueLength = 0 @@ -268,7 +269,7 @@ func (o *FileOutput) flush() { // Don't exit on panic defer func() { if r := recover(); r != nil { - log.Error().Stack().Msgf("PANIC while file flush: %v", r) + outputLogger.Error().Stack().Msgf("PANIC while file flush: %v", r) } }() @@ -285,7 +286,7 @@ func (o *FileOutput) flush() { if stat, err := o.file.Stat(); err == nil { o.currentFileSize = int(stat.Size()) } else { - log.Error().Err(err).Msgf("Error accessing file size") + outputLogger.Error().Err(err).Msgf("Error accessing file size") } } } diff --git a/pkg/file/output_s3.go b/pkg/file/output_s3.go index bec031c..8a0725d 100644 --- a/pkg/file/output_s3.go +++ b/pkg/file/output_s3.go @@ -19,6 +19,8 @@ import ( "github.com/rs/zerolog/log" ) +var s3Logger = log.With().Str("component", "s3").Logger() + // S3Output output plugin type S3Output struct { pathTemplate string @@ -32,7 +34,7 @@ type S3Output struct { // NewS3Output constructor for FileOutput, accepts path func NewS3Output(pathTemplate string, config *FileOutputConfig) *S3Output { if !pro.PRO { - log.Fatal().Msg("Using S3 output and input requires PRO license") + s3Logger.Fatal().Msg("Using S3 output and input requires PRO license") return nil } @@ -66,7 +68,7 @@ func NewS3Output(pathTemplate string, config *FileOutputConfig) *S3Output { func (o *S3Output) connect() { if o.session == nil { o.session = session.Must(session.NewSession(awsConfig())) - log.Info().Msg("[S3 Output] S3 connection successfully initialized") + s3Logger.Info().Msg("[S3 Output] S3 connection successfully initialized") } } @@ -113,7 +115,7 @@ func (o *S3Output) onBufferUpdate(path string) { file, err := os.Open(path) if err != nil { - log.Error().Err(err).Msgf("[S3 Output] Failed to open file %q", path) + s3Logger.Error().Err(err).Msgf("[S3 Output] Failed to open file %q", path) return } defer os.Remove(path) @@ -124,7 +126,7 @@ func (o *S3Output) onBufferUpdate(path string) { Key: aws.String(key), }) if err != nil { - log.Error().Err(err).Msgf("[S3 Output] Failed to upload data to %q/%q", bucket, key) + s3Logger.Error().Err(err).Msgf("[S3 Output] Failed to upload data to %q/%q", bucket, key) return } diff --git a/pkg/file/s3_reader.go b/pkg/file/s3_reader.go index fff42a0..8d28d42 100644 --- a/pkg/file/s3_reader.go +++ b/pkg/file/s3_reader.go @@ -11,8 +11,6 @@ import ( "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/s3" "github.com/buger/goreplay/pkg/pro" - - "github.com/rs/zerolog/log" ) // S3ReadCloser ... @@ -39,10 +37,10 @@ func awsConfig() *aws.Config { if endpoint := os.Getenv("AWS_ENDPOINT_URL"); endpoint != "" { config.Endpoint = aws.String(endpoint) - log.Debug().Msgf("Custom endpoint: %s", endpoint) + s3Logger.Debug().Msgf("Custom endpoint: %s", endpoint) } - log.Debug().Msgf("Connecting to S3. Region: %s", region) + s3Logger.Debug().Msgf("Connecting to S3. Region: %s", region) config.CredentialsChainVerboseErrors = aws.Bool(true) @@ -56,14 +54,14 @@ func awsConfig() *aws.Config { // NewS3ReadCloser returns new instance of S3 read closer func NewS3ReadCloser(path string) *S3ReadCloser { if !pro.PRO { - log.Fatal().Msg("Using S3 input and output require PRO license") + s3Logger.Fatal().Msg("Using S3 input and output require PRO license") return nil } bucket, key := parseS3Url(path) sess := session.Must(session.NewSession(awsConfig())) - log.Info().Msgf("S3 connection successfully initialized %v", path) + s3Logger.Info().Msgf("S3 connection successfully initialized %v", path) return &S3ReadCloser{ bucket: bucket, @@ -90,7 +88,7 @@ func (s *S3ReadCloser) Read(b []byte) (n int, e error) { resp, err := svc.GetObject(params) if err != nil { - log.Error().Err(err).Msgf("Error during getting file %s %s", s.bucket, s.key) + s3Logger.Error().Err(err).Msgf("Error during getting file %s %s", s.bucket, s.key) } else { s.totalSize, _ = strconv.Atoi(strings.Split(*resp.ContentRange, "/")[1]) s.buf.ReadFrom(resp.Body) diff --git a/pkg/http/input_http.go b/pkg/http/input_http.go index 7b78f6d..892ab7c 100644 --- a/pkg/http/input_http.go +++ b/pkg/http/input_http.go @@ -1,7 +1,6 @@ package http import ( - "log" "net" "net/http" "net/http/httputil" @@ -9,8 +8,12 @@ import ( "github.com/buger/goreplay/pkg/plugin" "github.com/buger/goreplay/pkg/proto" + + "github.com/rs/zerolog/log" ) +var inputLogger = log.With().Str("component", "input_http").Logger() + // HTTPInput used for sending requests to Gor via http type HTTPInput struct { data chan []byte @@ -67,14 +70,14 @@ func (i *HTTPInput) listen(address string) { i.listener, err = net.Listen("tcp", address) if err != nil { - log.Fatal("HTTP input listener failure:", err) + inputLogger.Fatal().Err(err).Msg("HTTP input listener failure") } i.address = i.listener.Addr().String() go func() { err = http.Serve(i.listener, mux) if err != nil && err != http.ErrServerClosed { - log.Fatal("HTTP input serve failure ", err) + inputLogger.Fatal().Err(err).Msg("HTTP input serve failure") } }() } diff --git a/pkg/http/output_http.go b/pkg/http/output_http.go index 6cc3af3..0ec852a 100644 --- a/pkg/http/output_http.go +++ b/pkg/http/output_http.go @@ -21,6 +21,8 @@ import ( "github.com/rs/zerolog/log" ) +var outputLogger = log.With().Str("component", "output_http").Logger() + const ( readChunkSize = 64 * 1024 ) @@ -92,7 +94,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) plugin.ReadWriter { newConfig := config.Copy() newConfig.url, err = url.Parse(address) if err != nil { - log.Fatal().Msg(fmt.Sprintf("[OUTPUT-HTTP] parse HTTP output URL error[%q]", err)) + outputLogger.Fatal().Msg(fmt.Sprintf("[OUTPUT-HTTP] parse HTTP output URL error[%q]", err)) } if newConfig.url.Scheme == "" { newConfig.url.Scheme = "http" @@ -244,7 +246,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *plugin.Message) { stop := time.Now() if err != nil { - log.Error().Err(err).Msg("[HTTP-OUTPUT]") + outputLogger.Error().Err(err).Msg("[HTTP-OUTPUT]") return } if resp == nil { @@ -286,12 +288,12 @@ func NewHTTPClient(config *HTTPOutputConfig) *HTTPClient { Timeout: client.config.Timeout, CheckRedirect: func(req *http.Request, via []*http.Request) error { if len(via) >= client.config.RedirectLimit { - log.Warn().Msgf("[HTTPCLIENT] maximum output-http-redirects[%d] reached!", client.config.RedirectLimit) + outputLogger.Warn().Msgf("[HTTPCLIENT] maximum output-http-redirects[%d] reached!", client.config.RedirectLimit) return http.ErrUseLastResponse } lastReq := via[len(via)-1] resp := req.Response - log.Info().Msgf("[HTTPCLIENT] HTTP redirects from %q to %q with %q", lastReq.Host, req.Host, resp.Status) + outputLogger.Info().Msgf("[HTTPCLIENT] HTTP redirects from %q to %q with %q", lastReq.Host, req.Host, resp.Status) return nil }, } diff --git a/pkg/pro/pro.go b/pkg/pro/pro.go index 54cd044..10d9b29 100644 --- a/pkg/pro/pro.go +++ b/pkg/pro/pro.go @@ -4,6 +4,7 @@ package pro // it must not be modified explicitly in production var PRO = false +// Enable enables PRO mode. Can be used ony in tests. func Enable() { PRO = true }