From e7059407bbdc45ea9c8bfe9370747582f165743e Mon Sep 17 00:00:00 2001 From: James Chao Date: Mon, 14 Sep 2026 16:00:50 -0700 Subject: [PATCH] add support to combine logger kafka and kafkaproducer to one line to reduce logs --- README.md | 4 ++ config/sample_webconfig.conf | 4 ++ go.mod | 24 +++---- go.sum | 60 +++++++++--------- http/webconfig_server.go | 114 ++++++++++++++++++++++++++++------ http/webconfig_server_test.go | 99 +++++++++++++++++++++++++++++ kafka/consumer.go | 42 ++++++++++--- kafka/consumer_group_test.go | 75 +++++++++++++++++++++- 8 files changed, 353 insertions(+), 69 deletions(-) diff --git a/README.md b/README.md index 7ef6617..4e46011 100644 --- a/README.md +++ b/README.md @@ -130,6 +130,10 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer ```shell kafka_producer { enabled = true + // one_line consolidates consumer and producer enqueue success under logger=kafka; + // two_line also emits the consumer processing success record and uses + // logger=kafkaproducer for producer records. + logging_mode = "one_line" brokers = "kafka-broker:9093" topic = "webconfig_downstream" diff --git a/config/sample_webconfig.conf b/config/sample_webconfig.conf index 04eebcb..65acc9c 100644 --- a/config/sample_webconfig.conf +++ b/config/sample_webconfig.conf @@ -406,6 +406,10 @@ webconfig { // forward kafka messages if needed kafka_producer { enabled = false + // Kafka success logging: "one_line" consolidates consumer and producer enqueue success + // under logger=kafka; "two_line" also emits the consumer processing success record and + // uses logger=kafkaproducer for producer records. + logging_mode = "one_line" brokers = "localhost:9092" topic = "webconfig_downstream" diff --git a/go.mod b/go.mod index 1a7ca88..47b6ec3 100644 --- a/go.mod +++ b/go.mod @@ -16,14 +16,14 @@ require ( github.com/twmb/murmur3 v1.1.8 github.com/vmihailenco/msgpack v4.0.4+incompatible github.com/vmihailenco/msgpack/v4 v4.3.12 - go.opentelemetry.io/otel v1.43.0 + go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0 - go.opentelemetry.io/otel/sdk v1.43.0 - go.opentelemetry.io/otel/trace v1.43.0 + go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/trace v1.44.0 go.uber.org/automaxprocs v1.6.0 go.uber.org/ratelimit v0.3.1 - golang.org/x/sync v0.20.0 + golang.org/x/sync v0.22.0 gotest.tools v2.2.0+incompatible modernc.org/sqlite v1.50.0 ) @@ -63,17 +63,17 @@ require ( github.com/vmihailenco/tagparser v0.1.2 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect - go.opentelemetry.io/otel/metric v1.43.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect - golang.org/x/crypto v0.52.0 // indirect - golang.org/x/net v0.55.0 // indirect - golang.org/x/sys v0.45.0 // indirect - golang.org/x/text v0.37.0 // indirect + golang.org/x/crypto v0.55.0 // indirect + golang.org/x/net v0.58.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect google.golang.org/appengine v1.6.8 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260504160031-60b97b32f348 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260504160031-60b97b32f348 // indirect - google.golang.org/grpc v1.82.1 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/grpc v1.83.2 // indirect google.golang.org/protobuf v1.36.11 // indirect gopkg.in/inf.v0 v0.9.1 // indirect modernc.org/libc v1.72.2 // indirect diff --git a/go.sum b/go.sum index 6af8102..5d4c236 100644 --- a/go.sum +++ b/go.sum @@ -138,22 +138,22 @@ github.com/vmihailenco/tagparser v0.1.2/go.mod h1:OeAg3pn3UbLjkWt+rN9oFYB6u/cQgq github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= -go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 h1:88Y4s2C8oTui1LGM6bTWkw0ICGcOLCAI5l6zsD1j20k= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0/go.mod h1:Vl1/iaggsuRlrHf/hfPJPvVag77kKyvrLeD10kpMl+A= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 h1:3iZJKlCZufyRzPzlQhUIWVmfltrXuGyfjREgGP3UUjc= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0/go.mod h1:/G+nUPfhq2e+qiXMGxMwumDrP5jtzU+mWN7/sjT2rak= go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0 h1:mS47AX77OtFfKG4vtp+84kuGSFZHTyxtXIN269vChY0= go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0/go.mod h1:PJnsC41lAGncJlPUniSwM81gc80GkgWJWr3cu2nKEtU= -go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= -go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= -go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= -go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= -go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= -go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= -go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= -go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw= @@ -169,11 +169,11 @@ go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= -golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= -golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= +golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= +golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= -golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM= -golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= +golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk= +golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40= golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= @@ -182,20 +182,20 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= -golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= -golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= @@ -205,13 +205,13 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= -golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= -golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= -golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= -golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= +golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE= +golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= @@ -219,12 +219,12 @@ gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc= google.golang.org/appengine v1.6.8 h1:IhEN5q69dyKagZPYMSdIjS2HqprW324FRQZJcGqPAsM= google.golang.org/appengine v1.6.8/go.mod h1:1jJ3jBArFh5pcgW8gCtRJnepW8FzD1V44FJffLiz/Ds= -google.golang.org/genproto/googleapis/api v0.0.0-20260504160031-60b97b32f348 h1:U8orV30l6KpDsi9dxU0CoJZGbjS8EEpw+6ba+XwGPQA= -google.golang.org/genproto/googleapis/api v0.0.0-20260504160031-60b97b32f348/go.mod h1:Yzdzr5OOZFgSsEV2D/Xi9NL3bszpXFAg0hFJiRohcD8= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260504160031-60b97b32f348 h1:pfIbyB44sWzHiCpRqIen67ZQnVXSfIxWrqUMk1qwODE= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260504160031-60b97b32f348/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= -google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= diff --git a/http/webconfig_server.go b/http/webconfig_server.go index 5277cd1..f403e63 100644 --- a/http/webconfig_server.go +++ b/http/webconfig_server.go @@ -74,6 +74,10 @@ const ( defaultSupplementaryAppendingEnabled = true authPrefixLength = 60 defaultMaxRequestBodyBytes = 1048576 + maxKafkaProducerLogPayloadBytes = 4096 + KafkaLoggingModeOneLine = "one_line" + KafkaLoggingModeTwoLine = "two_line" + defaultKafkaLoggingMode = KafkaLoggingModeOneLine ) var ( @@ -124,6 +128,7 @@ type WebconfigServer struct { supplementaryAppendingEnabled bool kafkaProducerEnabled bool kafkaProducerTopic string + kafkaLoggingMode string upstreamProfilesEnabled bool queryParamsValidationEnabled bool minTrust int @@ -306,6 +311,13 @@ func NewWebconfigServer(sc *common.ServerConfig, testOnly bool) *WebconfigServer var kafkaProducer sarama.AsyncProducer kafkaProducerEnabled := conf.GetBoolean("webconfig.kafka_producer.enabled") var kafkaProducerTopic string + kafkaLoggingMode := conf.GetString("webconfig.kafka_producer.logging_mode", defaultKafkaLoggingMode) + if kafkaLoggingMode != KafkaLoggingModeOneLine && kafkaLoggingMode != KafkaLoggingModeTwoLine { + if kafkaProducerEnabled { + panic(fmt.Errorf("webconfig.kafka_producer.logging_mode must be %q or %q", KafkaLoggingModeOneLine, KafkaLoggingModeTwoLine)) + } + kafkaLoggingMode = defaultKafkaLoggingMode + } if kafkaProducerEnabled { brokersStr := conf.GetString("webconfig.kafka_producer.brokers") if len(brokersStr) == 0 { @@ -414,6 +426,7 @@ func NewWebconfigServer(sc *common.ServerConfig, testOnly bool) *WebconfigServer supplementaryAppendingEnabled: supplementaryAppendingEnabled, kafkaProducerEnabled: kafkaProducerEnabled, kafkaProducerTopic: kafkaProducerTopic, + kafkaLoggingMode: kafkaLoggingMode, upstreamProfilesEnabled: upstreamProfilesEnabled, queryParamsValidationEnabled: queryParamsValidationEnabled, minTrust: minTrust, @@ -796,6 +809,17 @@ func (s *WebconfigServer) SetKafkaProducerTopic(x string) { s.kafkaProducerTopic = x } +func (s *WebconfigServer) KafkaLoggingMode() string { + return s.kafkaLoggingMode +} + +func (s *WebconfigServer) SetKafkaLoggingMode(mode string) { + if mode != KafkaLoggingModeOneLine && mode != KafkaLoggingModeTwoLine { + return + } + s.kafkaLoggingMode = mode +} + func (s *WebconfigServer) UpstreamProfilesEnabled() bool { return s.upstreamProfilesEnabled } @@ -973,6 +997,11 @@ func (s *WebconfigServer) logRequestStarts(w http.ResponseWriter, r *http.Reques mac = strings.ToUpper(mac) fields["cpe_mac"] = mac } + if r.Method == "POST" { + if subdocID, ok := params["subdoc_id"]; ok { + fields["subdoc_id"] = subdocID + } + } xwriter := NewXResponseWriter(w, time.Now(), token, fields) @@ -1145,8 +1174,11 @@ func GetResponseLogObjs(rbytes []byte) (interface{}, string) { return itf, "" } -func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMessage, fields log.Fields) { - tfields := common.CopyCoreLogFields(fields) +func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMessage, fields log.Fields, logMessage string) { + tfields := kafkaProducerLogFields(fields) + if _, ok := tfields["kafka_operation"]; !ok { + tfields["kafka_operation"] = "producer_send" + } bbytes, err := json.Marshal(m) if err != nil { @@ -1168,7 +1200,7 @@ func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMess if m := s.Metrics(); m != nil { m.ObserveKafkaProducerErr(s.KafkaProducerTopic(), -1) } - tfields["logger"] = "kafkaproducer" + tfields["logger"] = s.KafkaProducerLogger() tfields["error"] = r log.WithFields(tfields).Warn("dropped: producer closed during shutdown") } @@ -1176,16 +1208,32 @@ func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMess s.Input() <- outMessage - tfields["logger"] = "kafkaproducer" + tfields["logger"] = s.KafkaProducerLogger() tfields["output_topic"] = outMessage.Topic tfields["output_key"] = string(kbytes) - tfields["output_body"] = m - log.WithFields(tfields).Info("send") + log.WithFields(tfields).Info(logMessage + "; send") +} + +func kafkaProducerLogFields(fields log.Fields) log.Fields { + const producerFieldLimit = 15 + allowedFields := [...]string{ + "app_name", "audit_id", "body", "cpe_mac", "subdoc_id", + "kafka_key", "topic", "cluster_name", "kafka_partition", "kafka_offset", + "message_length", "duration", "event_name", "rpt", "kafka_operation", + } + producerFields := make(log.Fields, producerFieldLimit) + for _, field := range allowedFields { + if value, ok := fields[field]; ok { + producerFields[field] = common.FilterLogFields(log.Fields{field: value})[field] + } + } + return producerFields } func (s *WebconfigServer) ForwardSuccessKafkaMessages(messages []common.EventMessage, fields log.Fields) { - tfields := common.CopyCoreLogFields(fields) - tfields["logger"] = "kafkaproducer" + tfields := kafkaProducerLogFields(fields) + tfields["logger"] = s.KafkaProducerLogger() + tfields["kafka_operation"] = "state_correction_send" tfields["output_topic"] = s.KafkaProducerTopic() for _, m := range messages { @@ -1228,7 +1276,6 @@ func (s *WebconfigServer) ForwardSuccessKafkaMessages(messages []common.EventMes } tfields["output_key"] = mac - tfields["output_body"] = m log.WithFields(tfields).Info("send") } } @@ -1288,7 +1335,8 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) { continue } fields := make(log.Fields) - fields["logger"] = "kafkaproducer" + fields["logger"] = s.KafkaProducerLogger() + fields["kafka_operation"] = "producer_result" fields["output_topic"] = success.Topic fields["output_partition"] = success.Partition fields["output_offset"] = success.Offset @@ -1304,7 +1352,8 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) { m.ObserveKafkaProducerErr(pErr.Msg.Topic, pErr.Msg.Partition) } fields := make(log.Fields) - fields["logger"] = "kafkaproducer" + fields["logger"] = s.KafkaProducerLogger() + fields["kafka_operation"] = "producer_result" fields["output_topic"] = pErr.Msg.Topic fields["output_partition"] = pErr.Msg.Partition kbytes, err := pErr.Msg.Key.Encode() @@ -1318,14 +1367,7 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) { if err != nil { log.WithFields(fields).Error(common.NewError(err)) } else { - var itf interface{} - err1 := json.Unmarshal(vbytes, &itf) - if err1 != nil { - log.WithFields(fields).Error(common.NewError(err1)) - fields["output_body_text"] = base64.StdEncoding.EncodeToString(vbytes) - } else { - fields["output_body"] = itf - } + common.UpdateLogFields(fields, producerErrorPayloadFields(vbytes)) } log.WithFields(fields).Error(pErr.Err) @@ -1333,6 +1375,40 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) { } } +func (s *WebconfigServer) KafkaProducerLogger() string { + if s.KafkaLoggingMode() == KafkaLoggingModeTwoLine { + return "kafkaproducer" + } + return "kafka" +} + +func producerErrorPayloadFields(payload []byte) log.Fields { + fields := log.Fields{ + "output_body_bytes": len(payload), + "output_body_truncated": false, + } + maxRawBytes := (maxKafkaProducerLogPayloadBytes / 4) * 3 + if len(payload) <= maxRawBytes { + var body interface{} + if err := json.Unmarshal(payload, &body); err == nil { + fields["output_body"] = body + return fields + } + } + + fields["output_body_text"], fields["output_body_truncated"] = boundedKafkaPayload(payload) + return fields +} + +func boundedKafkaPayload(payload []byte) (string, bool) { + maxRawBytes := (maxKafkaProducerLogPayloadBytes / 4) * 3 + truncated := len(payload) > maxRawBytes + if truncated { + payload = payload[:maxRawBytes] + } + return base64.StdEncoding.EncodeToString(payload), truncated +} + func (s *WebconfigServer) StopXpcTracer() { sdkTraceProvider, ok := s.XpcTracer.OtelTracerProvider().(*sdktrace.TracerProvider) if ok && sdkTraceProvider != nil { diff --git a/http/webconfig_server_test.go b/http/webconfig_server_test.go index 6b611d4..832b8a0 100644 --- a/http/webconfig_server_test.go +++ b/http/webconfig_server_test.go @@ -47,6 +47,105 @@ func TestApiTokenAuthSecureDefaults(t *testing.T) { assert.Assert(t, deviceApiTokenAuthEnabledDefault, "device API token auth must default to enabled") } +func TestPostSubDocumentRequestIncludesSubdocID(t *testing.T) { + server := NewWebconfigServer(sc, true) + + req, err := http.NewRequest("POST", "/api/v1/device/001122334455/document/lan", nil) + assert.NilError(t, err) + req = mux.SetURLVars(req, map[string]string{ + "mac": "001122334455", + "subdoc_id": "lan", + }) + xw := server.logRequestStarts(httptest.NewRecorder(), req) + assert.Equal(t, xw.Audit()["subdoc_id"], "lan") + + getReq, err := http.NewRequest("GET", "/api/v1/device/001122334455/document/lan", nil) + assert.NilError(t, err) + getReq = mux.SetURLVars(getReq, map[string]string{ + "mac": "001122334455", + "subdoc_id": "lan", + }) + getXw := server.logRequestStarts(httptest.NewRecorder(), getReq) + _, ok := getXw.Audit()["subdoc_id"] + assert.Assert(t, !ok) +} + +func TestKafkaLoggingModeDefaultsToOneLine(t *testing.T) { + server := NewWebconfigServer(sc, true) + assert.Equal(t, server.KafkaLoggingMode(), KafkaLoggingModeOneLine) + + server.SetKafkaLoggingMode(KafkaLoggingModeTwoLine) + assert.Equal(t, server.KafkaLoggingMode(), KafkaLoggingModeTwoLine) + server.SetKafkaLoggingMode("invalid") + assert.Equal(t, server.KafkaLoggingMode(), KafkaLoggingModeTwoLine) +} + +func TestKafkaProducerLoggerFollowsLoggingMode(t *testing.T) { + server := NewWebconfigServer(sc, true) + assert.Equal(t, server.KafkaProducerLogger(), "kafka") + + server.SetKafkaLoggingMode(KafkaLoggingModeTwoLine) + assert.Equal(t, server.KafkaProducerLogger(), "kafkaproducer") +} + +func TestKafkaProducerLogFieldsAllowlist(t *testing.T) { + fields := kafkaProducerLogFields(log.Fields{ + "app_name": "webconfig", + "audit_id": "audit", + "cpe_mac": "001122334455", + "subdoc_id": "lan", + "event_name": "mqtt-get", + "header": map[string]string{"Authorization": "secret", "X-Test": "value"}, + "authorization": "secret", + }) + + _, hasHeader := fields["header"] + _, hasAuthorization := fields["authorization"] + assert.Assert(t, !hasHeader) + assert.Assert(t, !hasAuthorization) + assert.Equal(t, fields["subdoc_id"], "lan") + assert.Equal(t, fields["event_name"], "mqtt-get") + assert.Equal(t, fields["kafka_operation"], nil) +} + +func TestBoundedKafkaPayload(t *testing.T) { + payload := make([]byte, maxKafkaProducerLogPayloadBytes) + bounded, truncated := boundedKafkaPayload(payload) + assert.Equal(t, len(bounded), maxKafkaProducerLogPayloadBytes) + assert.Assert(t, truncated) + _, err := base64.StdEncoding.DecodeString(bounded) + assert.NilError(t, err) + + payload = append(payload, 'x') + bounded, truncated = boundedKafkaPayload(payload) + assert.Equal(t, len(bounded), maxKafkaProducerLogPayloadBytes) + assert.Assert(t, truncated) + _, err = base64.StdEncoding.DecodeString(bounded) + assert.NilError(t, err) + + payload = payload[:(maxKafkaProducerLogPayloadBytes/4)*3] + bounded, truncated = boundedKafkaPayload(payload) + assert.Assert(t, !truncated) + assert.Equal(t, len(bounded), maxKafkaProducerLogPayloadBytes) +} + +func TestProducerErrorPayloadFields(t *testing.T) { + fields := producerErrorPayloadFields([]byte(`{"status":"failed"}`)) + assert.Equal(t, fields["output_body"].(map[string]interface{})["status"], "failed") + _, hasText := fields["output_body_text"] + assert.Assert(t, !hasText) + assert.Assert(t, !fields["output_body_truncated"].(bool)) + + fields = producerErrorPayloadFields([]byte("not-json")) + _, hasBody := fields["output_body"] + assert.Assert(t, !hasBody) + _, hasText = fields["output_body_text"] + assert.Assert(t, hasText) + + fields = producerErrorPayloadFields(make([]byte, maxKafkaProducerLogPayloadBytes)) + assert.Assert(t, fields["output_body_truncated"].(bool)) +} + func TestConfigEndpointRemainsUnauthenticatedByDefault(t *testing.T) { server := NewWebconfigServer(sc, true) assert.Assert(t, !server.ConfigApiTokenAuthEnabled()) diff --git a/kafka/consumer.go b/kafka/consumer.go index adb95e6..90eea28 100644 --- a/kafka/consumer.go +++ b/kafka/consumer.go @@ -46,6 +46,30 @@ type Consumer struct { topicPartitionsMap map[string][]int32 } +func addKafkaEventLogFields(fields log.Fields, eventName, rptHeaderValue string) { + if eventName != "webpa-state" { + fields["event_name"] = eventName + } + if rptHeaderValue != "" { + fields["rpt"] = rptHeaderValue + } +} + +const maxKafkaMessageLogPayloadBytes = 4096 + +func boundedKafkaMessage(payload []byte) (string, bool) { + maxRawBytes := (maxKafkaMessageLogPayloadBytes / 4) * 3 + truncated := len(payload) > maxRawBytes + if truncated { + payload = payload[:maxRawBytes] + } + return base64.StdEncoding.EncodeToString(payload), truncated +} + +func shouldLogConsumerSuccess(producerEnabled bool, message *common.EventMessage, loggingMode string) bool { + return !producerEnabled || message == nil || loggingMode == wchttp.KafkaLoggingModeTwoLine +} + func NewConsumer(s *wchttp.WebconfigServer, ratelimitMessagesPerSecond int, m *common.AppMetrics, clusterName string, offsetEnum int64, topicPartitionsMap map[string][]int32) *Consumer { return &Consumer{ WebconfigServer: s, @@ -210,6 +234,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram messageLength := len(message.Value) fields := log.Fields{ "logger": "kafka", + "kafka_operation": "consumer_process", "app_name": c.AppName(), "kafka_lag": lag, "kafka_key": kafkaKey, @@ -244,8 +269,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram session.MarkMessage(message, "") duration := int(time.Since(start).Nanoseconds() / 1000000) fields["duration"] = duration - fields["event_name"] = eventName - fields["rpt"] = rptHeaderValue + addKafkaEventLogFields(fields, eventName, rptHeaderValue) forwardMessage := false if err != nil { @@ -255,12 +279,15 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram log.WithFields(fields).Trace("pending") } else { fields["error"] = err.Error() - fields["kafka_message"] = base64.StdEncoding.EncodeToString(message.Value) + fields["kafka_message"], fields["kafka_message_truncated"] = boundedKafkaMessage(message.Value) + fields["kafka_message_bytes"] = len(message.Value) log.WithFields(fields).Error("errors") } } else { forwardMessage = true - log.WithFields(fields).Info(logMessage) + if shouldLogConsumerSuccess(c.KafkaProducerEnabled(), m, c.KafkaLoggingMode()) { + log.WithFields(fields).Info(logMessage) + } } // build metrics dimensions and update metrics @@ -281,7 +308,8 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram } if c.KafkaProducerEnabled() && m != nil && forwardMessage { - c.ForwardKafkaMessage(message.Key, m, fields) + fields["kafka_operation"] = "producer_send" + c.ForwardKafkaMessage(message.Key, m, fields, logMessage) if len(m.Reports) == 0 { if m.HttpStatusCode != nil && *m.HttpStatusCode == http.StatusNotModified && len(updatedSubdocIds) > 0 { // build a root/success message @@ -294,7 +322,8 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram TransactionUuid: m.TransactionUuid, Version: m.Version, } - c.ForwardKafkaMessage(message.Key, em, fields) + fields["kafka_operation"] = "state_correction_send" + c.ForwardKafkaMessage(message.Key, em, fields, logMessage) } } } @@ -303,7 +332,6 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram return nil } } - return nil } func (c *Consumer) AppName() string { diff --git a/kafka/consumer_group_test.go b/kafka/consumer_group_test.go index d720eb2..c1fcd78 100644 --- a/kafka/consumer_group_test.go +++ b/kafka/consumer_group_test.go @@ -14,17 +14,90 @@ * limitations under the License. * * SPDX-License-Identifier: Apache-2.0 -*/ + */ package kafka import ( + "encoding/base64" "testing" "time" "github.com/IBM/sarama" + "github.com/rdkcentral/webconfig/common" + wchttp "github.com/rdkcentral/webconfig/http" + log "github.com/sirupsen/logrus" "gotest.tools/assert" ) +func TestKafkaSuccessLoggingMode(t *testing.T) { + message := &common.EventMessage{} + tests := []struct { + name string + producerEnabled bool + message *common.EventMessage + loggingMode string + want bool + }{ + {name: "producer disabled", want: true}, + {name: "one line", producerEnabled: true, message: message, loggingMode: wchttp.KafkaLoggingModeOneLine, want: false}, + {name: "two line", producerEnabled: true, message: message, loggingMode: wchttp.KafkaLoggingModeTwoLine, want: true}, + {name: "no message", producerEnabled: true, loggingMode: wchttp.KafkaLoggingModeOneLine, want: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, shouldLogConsumerSuccess(test.producerEnabled, test.message, test.loggingMode), test.want) + }) + } +} + +func TestBoundedKafkaMessage(t *testing.T) { + payload := make([]byte, maxKafkaMessageLogPayloadBytes+1) + encoded, truncated := boundedKafkaMessage(payload) + assert.Assert(t, truncated) + assert.Equal(t, len(encoded), maxKafkaMessageLogPayloadBytes) + _, err := base64.StdEncoding.DecodeString(encoded) + assert.NilError(t, err) + + payload = payload[:(maxKafkaMessageLogPayloadBytes/4)*3] + encoded, truncated = boundedKafkaMessage(payload) + assert.Assert(t, !truncated) + assert.Equal(t, len(encoded), maxKafkaMessageLogPayloadBytes) +} + +func TestKafkaEventLogFields(t *testing.T) { + tests := []struct { + name string + eventName string + rpt string + wantEvent string + wantRPT string + hasEvent bool + hasRPT bool + }{ + {name: "webpa state default", eventName: "webpa-state"}, + {name: "mqtt get", eventName: "mqtt-get", rpt: "x/fr/webconfig/get", wantEvent: "mqtt-get", wantRPT: "x/fr/webconfig/get", hasEvent: true, hasRPT: true}, + {name: "mqtt state", eventName: "mqtt-state", rpt: "x/fr/webconfig/poke", wantEvent: "mqtt-state", wantRPT: "x/fr/webconfig/poke", hasEvent: true, hasRPT: true}, + {name: "unknown event without rpt", eventName: "unknown-no-rpt", wantEvent: "unknown-no-rpt", hasEvent: true}, + {name: "unknown event with rpt", eventName: "unknown-rpt", rpt: "indigo", wantEvent: "unknown-rpt", wantRPT: "indigo", hasEvent: true, hasRPT: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fields := log.Fields{} + addKafkaEventLogFields(fields, test.eventName, test.rpt) + value, hasEvent := fields["event_name"] + assert.Equal(t, hasEvent, test.hasEvent) + if test.hasEvent { + assert.Equal(t, value, test.wantEvent) + } + value, hasRPT := fields["rpt"] + assert.Equal(t, hasRPT, test.hasRPT) + if test.hasRPT { + assert.Equal(t, value, test.wantRPT) + } + }) + } +} + func TestGetEventName(t *testing.T) { // ==== mqtt-get ==== rheader := &sarama.RecordHeader{