conn: change design log

Author: New year

connection.go

@@ -32,7 +32,6 @@ const (
 const shutdownEventKey = "box.shutdown"
 
 type ConnEventKind int
-type ConnLogKind int
 
 var (
 	errUnknownRequest = errors.New("the passed connected request doesn't belong " +
@@ -50,19 +49,6 @@ const (
 	Shutdown
 	// Either reconnect attempts exhausted, or explicit Close is called.
 	Closed
-
-	// LogReconnectFailed is logged when reconnect attempt failed.
-	//	LogReconnectFailed ConnLogKind = iota + 1
-	// LogLastReconnectFailed is logged when last reconnect attempt failed,
-	// connection will be closed after that.
-	//	LogLastReconnectFailed
-	// LogUnexpectedResultId is logged when response with unknown id was received.
-	// Most probably it is due to request timeout.
-	LogUnexpectedResultId
-	// LogWatchEventReadFailed is logged when failed to read a watch event.
-//	LogWatchEventReadFailed
-// LogBoxSessionPushUnsupported is logged when response type turned IPROTO_CHUNK.
-//	LogBoxSessionPushUnsupported
 )
 
 // ConnEvent is sent throw Notify channel specified in Opts.
@@ -80,126 +66,6 @@ type connWatchEvent struct {
 
 var epoch = time.Now()
 
-type ConnLogEvent interface{}
-
-// Events of logger
-type LogReconnectFailed struct {
-	Reconnects    uint
-	MaxReconnects uint
-	Err           error
-}
-
-type LogLastReconnectFailed struct {
-	Err error
-}
-type LogUnexpectedResultID struct {
-	Header Header
-}
-type LogWatchEventReadFailed struct {
-	Err error
-}
-type LogBoxSessionPushUnsupported struct {
-	Header Header
-}
-
-// Logger is logger type expected to be passed in options.
-type Logger interface {
-	Report(event ConnLogEvent, conn *Connection)
-}
-
-type defaultSlogLogger struct {
-	l *slog.Logger
-}
-
-func NewDefaultSlogLogger(l *slog.Logger) Logger {
-	if l == nil {
-		l = slog.Default()
-	}
-	return &defaultSlogLogger{l: l}
-}
-
-type defaultLogger struct{}
-
-func attrsToAny(attrs []slog.Attr) []any {
-	out := make([]any, 0, len(attrs)*2)
-	for _, a := range attrs {
-		out = append(out, a.Key, a.Value.Any())
-	}
-	return out
-}
-
-func (d *defaultSlogLogger) Report(event ConnLogEvent, conn *Connection) {
-	level, msg, attrs := eventToSlog(event, conn)
-	args := attrsToAny(attrs)
-	ctx := context.Background()
-	switch level {
-	case slog.LevelError:
-		d.l.Error(msg, args...)
-	case slog.LevelWarn:
-		d.l.Warn(msg, args...)
-	case slog.LevelInfo:
-		d.l.Info(msg, args...)
-	case slog.LevelDebug:
-		d.l.Debug(msg, args...)
-	default:
-		d.l.Log(ctx, level, msg, args...)
-	}
-}
-
-func eventToSlog(event ConnLogEvent, conn *Connection) (slog.Level, string, []slog.Attr) {
-	addr := "<nil>"
-	if conn != nil && conn.Addr() != nil {
-		addr = fmt.Sprintf("%s", conn.Addr())
-	}
-
-	switch e := event.(type) {
-	case LogReconnectFailed:
-		msg := "tarantool: reconnect failed"
-		attrs := []slog.Attr{
-			slog.Int("reconnects", int(e.Reconnects)),
-			slog.Int("max_reconnects", int(e.MaxReconnects)),
-			slog.String("addr", addr),
-		}
-		if e.Err != nil {
-			attrs = append(attrs, slog.String("error", e.Err.Error()))
-		}
-		return slog.LevelWarn, msg, attrs
-
-	case LogLastReconnectFailed:
-		msg := "tarantool: last reconnect failed, giving it up"
-		attrs := []slog.Attr{slog.String("addr", addr)}
-		if e.Err != nil {
-			attrs = append(attrs, slog.String("error", e.Err.Error()))
-		}
-		return slog.LevelError, msg, attrs
-
-	case LogUnexpectedResultID:
-		msg := "tarantool: unexpected response request id (probably cancelled request)"
-		attrs := []slog.Attr{
-			slog.String("addr", addr),
-			slog.Any("request_id", e.Header.RequestId),
-		}
-		return slog.LevelWarn, msg, attrs
-
-	case LogWatchEventReadFailed:
-		msg := "tarantool: unable to parse watch event"
-		attrs := []slog.Attr{}
-		if e.Err != nil {
-			attrs = append(attrs, slog.String("error", e.Err.Error()))
-		}
-		return slog.LevelWarn, msg, attrs
-
-	case LogBoxSessionPushUnsupported:
-		msg := "tarantool: unsupported box.session.push()"
-		attrs := []slog.Attr{slog.Any("request_id", e.Header.RequestId), slog.String("addr", addr)}
-		return slog.LevelInfo, msg, attrs
-
-	default:
-		// unknown event: log type and value
-		return slog.LevelInfo, "tarantool: unexpected event", []slog.Attr{slog.Any("event", event), slog.String("addr", addr)}
-	}
-}
-
 // Connection is a handle with a single connection to a Tarantool instance.
 //
 // It is created and configured with Connect function, and could not be
@@ -442,7 +308,7 @@ func Connect(ctx context.Context, dialer Dialer, opts Opts) (conn *Connection, e
 	}
 
 	if conn.opts.Logger == nil {
-		conn.opts.Logger = defaultLogger{}
+		conn.opts.Logger = NewSlogLogger(slog.Default())
 	}
 
 	conn.cond = sync.NewCond(&conn.mutex)
@@ -483,6 +349,29 @@ func Connect(ctx context.Context, dialer Dialer, opts Opts) (conn *Connection, e
 	return conn, err
 }
 
+func (conn *Connection) logEvent(event LogEvent) {
+	if conn.opts.Logger != nil {
+		conn.opts.Logger.Report(event, conn)
+	}
+}
+
+// stateToString преобразует состояние соединения в строку
+func (conn *Connection) stateToString() string {
+	state := atomic.LoadUint32(&conn.state)
+	switch state {
+	case connDisconnected:
+		return "disconnected"
+	case connConnected:
+		return "connected"
+	case connShutdown:
+		return "shutdown"
+	case connClosed:
+		return "closed"
+	default:
+		return "unknown"
+	}
+}
+
 // ConnectedNow reports if connection is established at the moment.
 func (conn *Connection) ConnectedNow() bool {
 	return atomic.LoadUint32(&conn.state) == connConnected
@@ -652,10 +541,20 @@ func pack(h *smallWBuf, enc *msgpack.Encoder, reqid uint32,
 	return
 }
 
+// connect обновленная версия с новым логированием
 func (conn *Connection) connect(ctx context.Context) error {
 	var err error
 	if conn.c == nil && conn.state == connDisconnected {
 		if err = conn.dial(ctx); err == nil {
+			// Определяем номер попытки переподключения
+			var reconnects uint = 0
+			//TODO нужно отслеживать количество переподключений
+
+			// Логируем успешное подключение
+			conn.logEvent(ConnectedEvent{
+				baseEvent:  newBaseEvent(conn.addr),
+				Reconnects: reconnects,
+			})
 			conn.notify(Connected)
 			return nil
 		}
@@ -666,24 +565,30 @@ func (conn *Connection) connect(ctx context.Context) error {
 	return err
 }
 
+// closeConnection обновленная версия с новым логированием
 func (conn *Connection) closeConnection(neterr error, forever bool) (err error) {
 	conn.lockShards()
 	defer conn.unlockShards()
+
 	if forever {
 		if conn.state != connClosed {
 			close(conn.control)
 			atomic.StoreUint32(&conn.state, connClosed)
 			conn.cond.Broadcast()
-			// Free the resources.
-			if conn.shutdownWatcher != nil {
-				go conn.shutdownWatcher.Unregister()
-				conn.shutdownWatcher = nil
-			}
+			// Логируем закрытие соединения
+			conn.logEvent(ClosedEvent{
+				baseEvent: newBaseEvent(conn.addr),
+			})
 			conn.notify(Closed)
 		}
 	} else {
 		atomic.StoreUint32(&conn.state, connDisconnected)
 		conn.cond.Broadcast()
+		// Логируем отключение
+		conn.logEvent(DisconnectedEvent{
+			baseEvent: newBaseEvent(conn.addr),
+			Reason:    neterr,
+		})
 		conn.notify(Disconnected)
 	}
 	if conn.c != nil {
@@ -715,6 +620,7 @@ func (conn *Connection) getDialTimeout() time.Duration {
 	return dialTimeout
 }
 
+// runReconnects обновленная версия с новым логированием
 func (conn *Connection) runReconnects(ctx context.Context) error {
 	dialTimeout := conn.getDialTimeout()
 	var reconnects uint
@@ -728,14 +634,6 @@ func (conn *Connection) runReconnects(ctx context.Context) error {
 		cancel()
 
 		if err != nil {
-			// The error will most likely be the one that Dialer
-			// returns to us due to the context being cancelled.
-			// Although this is not guaranteed. For example,
-			// if the dialer may throw another error before checking
-			// the context, and the context has already been
-			// canceled. Or the context was not canceled after
-			// the error was thrown, but before the context was
-			// checked here.
 			if ctx.Err() != nil {
 				return err
 			}
@@ -747,23 +645,33 @@ func (conn *Connection) runReconnects(ctx context.Context) error {
 			return nil
 		}
 
-		conn.opts.Logger.Report(LogReconnectFailed, conn, reconnects, err)
+		// Новое логирование события
+		conn.logEvent(ReconnectFailedEvent{
+			baseEvent:     newBaseEvent(conn.addr),
+			Reconnects:    reconnects,
+			MaxReconnects: conn.opts.MaxReconnects,
+			Error:         err,
+			IsInitial:     conn.addr == nil,
+		})
+
 		conn.notify(ReconnectFailed)
 		reconnects++
 		conn.mutex.Unlock()
 
 		select {
 		case <-ctx.Done():
-			// Since the context is cancelled, we don't need to do anything.
-			// Conn.connect() will return the correct error.
 		case <-t.C:
 		}
 
 		conn.mutex.Lock()
 	}
 
-	conn.opts.Logger.Report(LogLastReconnectFailed, conn, err)
-	// mark connection as closed to avoid reopening by another goroutine
+	// Новое логирование события последней неудачной попытки
+	conn.logEvent(LastReconnectFailedEvent{
+		baseEvent: newBaseEvent(conn.addr),
+		Error:     err,
+	})
+
 	return ClientError{ErrConnectionClosed, "last reconnect failed"}
 }
 
@@ -917,6 +825,7 @@ func readWatchEvent(reader io.Reader) (connWatchEvent, error) {
 	return event, nil
 }
 
+// reader обновленная версия с новым логированием
 func (conn *Connection) reader(r io.Reader, c Conn) {
 	events := make(chan connWatchEvent, 1024)
 	defer close(events)
@@ -953,11 +862,19 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 					ErrProtocolError,
 					fmt.Sprintf("failed to decode IPROTO_EVENT: %s", err),
 				}
-				conn.opts.Logger.Report(LogWatchEventReadFailed, conn, err)
+				// Новое логирование события
+				conn.logEvent(WatchEventReadFailedEvent{
+					baseEvent: newBaseEvent(conn.addr),
+					Error:     err,
+				})
 			}
 			continue
 		} else if code == iproto.IPROTO_CHUNK {
-			conn.opts.Logger.Report(LogBoxSessionPushUnsupported, conn, header)
+			// Новое логирование события
+			conn.logEvent(BoxSessionPushUnsupportedEvent{
+				baseEvent: newBaseEvent(conn.addr),
+				RequestId: header.RequestId,
+			})
 		} else {
 			if fut = conn.fetchFuture(header.RequestId); fut != nil {
 				if err := fut.SetResponse(header, &buf); err != nil {
@@ -968,7 +885,11 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 		}
 
 		if fut == nil {
-			conn.opts.Logger.Report(LogUnexpectedResultId, conn, header)
+			// Новое логирование события
+			conn.logEvent(UnexpectedResultIdEvent{
+				baseEvent: newBaseEvent(conn.addr),
+				RequestId: header.RequestId,
+			})
 		}
 	}
 }
@@ -1214,6 +1135,7 @@ func (conn *Connection) getFutureImp(reqid uint32, fetch bool) *Future {
 	}
 }
 
+// timeouts обновленная версия с новым логированием
 func (conn *Connection) timeouts() {
 	timeout := conn.opts.Timeout
 	t := time.NewTimer(timeout)
@@ -1247,6 +1169,13 @@ func (conn *Connection) timeouts() {
 					})
 					conn.markDone(fut)
 					shard.bufmut.Unlock()
+
+					// Логируем таймаут
+					conn.logEvent(TimeoutEvent{
+						baseEvent: newBaseEvent(conn.addr),
+						RequestId: fut.requestId,
+						Timeout:   timeout,
+					})
 				}
 				if pair.first != nil && pair.first.timeout < minNext {
 					minNext = pair.first.timeout