From d9d5a01889a1bc1cfdc99b1b68447996bb45cf1d Mon Sep 17 00:00:00 2001 From: Francesco Gualazzi Date: Mon, 12 Aug 2024 15:05:21 +0200 Subject: [PATCH] Kafka: add hook to log connection error reasons (#541) The current `metricsHook` struct that collects metrics based on Kafka hooks is not equipped to also report details of _why_ connection errors are caused. --- Details We provide a new struct that implements a (single, for now) kgo Hook dedicated to logging errors during the broker connection phase. The struct can be iterated in the future to implement other monitoring hooks, providing logging utilities where the metrics would have too high cardinality. The proposed design adds the new logger hook behind the `CommonConfig.DisableTelemetry` flag (similarly to the metrics hook). This can be improved and it is not a perfect solution, but it has to be done this way until a bigger refactor is done to expose the `CommonConfig.hook` fieldin some way. The more idiomatic, but also more expensive, solution would be to allow exporting franz-go logs into the Logger. This is mentioned in https://github.com/elastic/apm-queue/issues/540 --------- Signed-off-by: inge4pres Co-authored-by: Edoardo Tenani <526307+endorama@users.noreply.github.com> --- kafka/common.go | 2 +- kafka/logger.go | 46 ++++++++++++++++++++++++++++++++++ kafka/logger_test.go | 59 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 1 deletion(-) create mode 100644 kafka/logger.go create mode 100644 kafka/logger_test.go diff --git a/kafka/common.go b/kafka/common.go index 7c771fe7..d8e9b3ef 100644 --- a/kafka/common.go +++ b/kafka/common.go @@ -288,7 +288,7 @@ func (cfg *CommonConfig) newClient(topicAttributeFunc TopicAttributeFunc, additi return nil, fmt.Errorf("kafka: failed creating kgo metrics hooks: %w", err) } opts = append(opts, - kgo.WithHooks(metricHooks), + kgo.WithHooks(metricHooks, &loggerHook{logger: cfg.Logger}), ) } if cfg.MetadataMaxAge > 0 { diff --git a/kafka/logger.go b/kafka/logger.go new file mode 100644 index 00000000..d233cc1a --- /dev/null +++ b/kafka/logger.go @@ -0,0 +1,46 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package kafka + +import ( + "net" + "time" + + "github.com/twmb/franz-go/pkg/kgo" + "go.uber.org/zap" +) + +// Compile-time check that loggerHook implements the HookBrokerConnect interface. +var _ kgo.HookBrokerConnect = new(loggerHook) + +type loggerHook struct { + logger *zap.Logger +} + +// OnBrokerConnect implements the kgo.HookBrokerConnect interface. +func (l *loggerHook) OnBrokerConnect(meta kgo.BrokerMetadata, dialDur time.Duration, _ net.Conn, err error) { + if err != nil { + l.logger.Error("failed to connect to broker", + zap.Error(err), + zap.String("duration", dialDur.String()), + zap.String("host", meta.Host), + zap.Int32("port", meta.Port), + zap.Stack("stack"), + ) + } +} diff --git a/kafka/logger_test.go b/kafka/logger_test.go new file mode 100644 index 00000000..5061bc8a --- /dev/null +++ b/kafka/logger_test.go @@ -0,0 +1,59 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package kafka + +import ( + "context" + "errors" + "net" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" +) + +func TestHookLogsFailedDial(t *testing.T) { + cluster, cfg := newFakeCluster(t) + t.Cleanup(cluster.Close) + + core, logs := observer.New(zap.ErrorLevel) + cfg.Logger = zap.New(core) + // Simulate returning an error when dialing the broker. + const errorMsg = "busted" + cfg.Dialer = func(context.Context, string, string) (net.Conn, error) { + return nil, errors.New(errorMsg) + } + + // Calling newClient triggers the metadata refresh, forcing a connection to the fake cluster + // using the broken dialer. + c, err := cfg.newClient(func(string) attribute.KeyValue { return attribute.String("k", "v") }) + require.NoError(t, err) + assert.Error(t, c.Ping(context.Background())) + + observedLogs := logs.FilterMessage("failed to connect to broker").TakeAll() + // Franz-go will retry once to connect to the broker, so we might see either one or two log lines. + assert.True(t, len(observedLogs) == 1 || len(observedLogs) == 2, + "expected one or two log lines, got %#v", observedLogs) + + // The error message should contain the error message from the dialer. + assert.EqualValues(t, observedLogs[0].ContextMap()["error"], errorMsg) + assert.Contains(t, observedLogs[0].ContextMap(), "duration") +}