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 #540

---------

Signed-off-by: inge4pres <francesco.gualazzi@elastic.co>
Co-authored-by: Edoardo Tenani <526307+endorama@users.noreply.github.com>

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 {

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"),
+		)
+	}
+}

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")
+}