method_logger.go

Documentation: google.golang.org/grpc/internal/binarylog

     1  /*
     2   *
     3   * Copyright 2018 gRPC authors.
     4   *
     5   * Licensed under the Apache License, Version 2.0 (the "License");
     6   * you may not use this file except in compliance with the License.
     7   * You may obtain a copy of the License at
     8   *
     9   *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   */
    18  
    19  package binarylog
    20  
    21  import (
    22  	"context"
    23  	"net"
    24  	"strings"
    25  	"sync/atomic"
    26  	"time"
    27  
    28  	binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
    29  	"google.golang.org/grpc/metadata"
    30  	"google.golang.org/grpc/status"
    31  	"google.golang.org/protobuf/proto"
    32  	"google.golang.org/protobuf/types/known/durationpb"
    33  	"google.golang.org/protobuf/types/known/timestamppb"
    34  )
    35  
    36  type callIDGenerator struct {
    37  	id uint64
    38  }
    39  
    40  func (g *callIDGenerator) next() uint64 {
    41  	id := atomic.AddUint64(&g.id, 1)
    42  	return id
    43  }
    44  
    45  // reset is for testing only, and doesn't need to be thread safe.
    46  func (g *callIDGenerator) reset() {
    47  	g.id = 0
    48  }
    49  
    50  var idGen callIDGenerator
    51  
    52  // MethodLogger is the sub-logger for each method.
    53  //
    54  // This is used in the 1.0 release of gcp/observability, and thus must not be
    55  // deleted or changed.
    56  type MethodLogger interface {
    57  	Log(context.Context, LogEntryConfig)
    58  }
    59  
    60  // TruncatingMethodLogger is a method logger that truncates headers and messages
    61  // based on configured fields.
    62  type TruncatingMethodLogger struct {
    63  	headerMaxLen, messageMaxLen uint64
    64  
    65  	callID          uint64
    66  	idWithinCallGen *callIDGenerator
    67  
    68  	sink Sink // TODO(blog): make this pluggable.
    69  }
    70  
    71  // NewTruncatingMethodLogger returns a new truncating method logger.
    72  //
    73  // This is used in the 1.0 release of gcp/observability, and thus must not be
    74  // deleted or changed.
    75  func NewTruncatingMethodLogger(h, m uint64) *TruncatingMethodLogger {
    76  	return &TruncatingMethodLogger{
    77  		headerMaxLen:  h,
    78  		messageMaxLen: m,
    79  
    80  		callID:          idGen.next(),
    81  		idWithinCallGen: &callIDGenerator{},
    82  
    83  		sink: DefaultSink, // TODO(blog): make it pluggable.
    84  	}
    85  }
    86  
    87  // Build is an internal only method for building the proto message out of the
    88  // input event. It's made public to enable other library to reuse as much logic
    89  // in TruncatingMethodLogger as possible.
    90  func (ml *TruncatingMethodLogger) Build(c LogEntryConfig) *binlogpb.GrpcLogEntry {
    91  	m := c.toProto()
    92  	timestamp := timestamppb.Now()
    93  	m.Timestamp = timestamp
    94  	m.CallId = ml.callID
    95  	m.SequenceIdWithinCall = ml.idWithinCallGen.next()
    96  
    97  	switch pay := m.Payload.(type) {
    98  	case *binlogpb.GrpcLogEntry_ClientHeader:
    99  		m.PayloadTruncated = ml.truncateMetadata(pay.ClientHeader.GetMetadata())
   100  	case *binlogpb.GrpcLogEntry_ServerHeader:
   101  		m.PayloadTruncated = ml.truncateMetadata(pay.ServerHeader.GetMetadata())
   102  	case *binlogpb.GrpcLogEntry_Message:
   103  		m.PayloadTruncated = ml.truncateMessage(pay.Message)
   104  	}
   105  	return m
   106  }
   107  
   108  // Log creates a proto binary log entry, and logs it to the sink.
   109  func (ml *TruncatingMethodLogger) Log(ctx context.Context, c LogEntryConfig) {
   110  	ml.sink.Write(ml.Build(c))
   111  }
   112  
   113  func (ml *TruncatingMethodLogger) truncateMetadata(mdPb *binlogpb.Metadata) (truncated bool) {
   114  	if ml.headerMaxLen == maxUInt {
   115  		return false
   116  	}
   117  	var (
   118  		bytesLimit = ml.headerMaxLen
   119  		index      int
   120  	)
   121  	// At the end of the loop, index will be the first entry where the total
   122  	// size is greater than the limit:
   123  	//
   124  	// len(entry[:index]) <= ml.hdr && len(entry[:index+1]) > ml.hdr.
   125  	for ; index < len(mdPb.Entry); index++ {
   126  		entry := mdPb.Entry[index]
   127  		if entry.Key == "grpc-trace-bin" {
   128  			// "grpc-trace-bin" is a special key. It's kept in the log entry,
   129  			// but not counted towards the size limit.
   130  			continue
   131  		}
   132  		currentEntryLen := uint64(len(entry.GetKey())) + uint64(len(entry.GetValue()))
   133  		if currentEntryLen > bytesLimit {
   134  			break
   135  		}
   136  		bytesLimit -= currentEntryLen
   137  	}
   138  	truncated = index < len(mdPb.Entry)
   139  	mdPb.Entry = mdPb.Entry[:index]
   140  	return truncated
   141  }
   142  
   143  func (ml *TruncatingMethodLogger) truncateMessage(msgPb *binlogpb.Message) (truncated bool) {
   144  	if ml.messageMaxLen == maxUInt {
   145  		return false
   146  	}
   147  	if ml.messageMaxLen >= uint64(len(msgPb.Data)) {
   148  		return false
   149  	}
   150  	msgPb.Data = msgPb.Data[:ml.messageMaxLen]
   151  	return true
   152  }
   153  
   154  // LogEntryConfig represents the configuration for binary log entry.
   155  //
   156  // This is used in the 1.0 release of gcp/observability, and thus must not be
   157  // deleted or changed.
   158  type LogEntryConfig interface {
   159  	toProto() *binlogpb.GrpcLogEntry
   160  }
   161  
   162  // ClientHeader configs the binary log entry to be a ClientHeader entry.
   163  type ClientHeader struct {
   164  	OnClientSide bool
   165  	Header       metadata.MD
   166  	MethodName   string
   167  	Authority    string
   168  	Timeout      time.Duration
   169  	// PeerAddr is required only when it's on server side.
   170  	PeerAddr net.Addr
   171  }
   172  
   173  func (c *ClientHeader) toProto() *binlogpb.GrpcLogEntry {
   174  	// This function doesn't need to set all the fields (e.g. seq ID). The Log
   175  	// function will set the fields when necessary.
   176  	clientHeader := &binlogpb.ClientHeader{
   177  		Metadata:   mdToMetadataProto(c.Header),
   178  		MethodName: c.MethodName,
   179  		Authority:  c.Authority,
   180  	}
   181  	if c.Timeout > 0 {
   182  		clientHeader.Timeout = durationpb.New(c.Timeout)
   183  	}
   184  	ret := &binlogpb.GrpcLogEntry{
   185  		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_HEADER,
   186  		Payload: &binlogpb.GrpcLogEntry_ClientHeader{
   187  			ClientHeader: clientHeader,
   188  		},
   189  	}
   190  	if c.OnClientSide {
   191  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
   192  	} else {
   193  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
   194  	}
   195  	if c.PeerAddr != nil {
   196  		ret.Peer = addrToProto(c.PeerAddr)
   197  	}
   198  	return ret
   199  }
   200  
   201  // ServerHeader configs the binary log entry to be a ServerHeader entry.
   202  type ServerHeader struct {
   203  	OnClientSide bool
   204  	Header       metadata.MD
   205  	// PeerAddr is required only when it's on client side.
   206  	PeerAddr net.Addr
   207  }
   208  
   209  func (c *ServerHeader) toProto() *binlogpb.GrpcLogEntry {
   210  	ret := &binlogpb.GrpcLogEntry{
   211  		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_HEADER,
   212  		Payload: &binlogpb.GrpcLogEntry_ServerHeader{
   213  			ServerHeader: &binlogpb.ServerHeader{
   214  				Metadata: mdToMetadataProto(c.Header),
   215  			},
   216  		},
   217  	}
   218  	if c.OnClientSide {
   219  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
   220  	} else {
   221  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
   222  	}
   223  	if c.PeerAddr != nil {
   224  		ret.Peer = addrToProto(c.PeerAddr)
   225  	}
   226  	return ret
   227  }
   228  
   229  // ClientMessage configs the binary log entry to be a ClientMessage entry.
   230  type ClientMessage struct {
   231  	OnClientSide bool
   232  	// Message can be a proto.Message or []byte. Other messages formats are not
   233  	// supported.
   234  	Message any
   235  }
   236  
   237  func (c *ClientMessage) toProto() *binlogpb.GrpcLogEntry {
   238  	var (
   239  		data []byte
   240  		err  error
   241  	)
   242  	if m, ok := c.Message.(proto.Message); ok {
   243  		data, err = proto.Marshal(m)
   244  		if err != nil {
   245  			grpclogLogger.Infof("binarylogging: failed to marshal proto message: %v", err)
   246  		}
   247  	} else if b, ok := c.Message.([]byte); ok {
   248  		data = b
   249  	} else {
   250  		grpclogLogger.Infof("binarylogging: message to log is neither proto.message nor []byte")
   251  	}
   252  	ret := &binlogpb.GrpcLogEntry{
   253  		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_MESSAGE,
   254  		Payload: &binlogpb.GrpcLogEntry_Message{
   255  			Message: &binlogpb.Message{
   256  				Length: uint32(len(data)),
   257  				Data:   data,
   258  			},
   259  		},
   260  	}
   261  	if c.OnClientSide {
   262  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
   263  	} else {
   264  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
   265  	}
   266  	return ret
   267  }
   268  
   269  // ServerMessage configs the binary log entry to be a ServerMessage entry.
   270  type ServerMessage struct {
   271  	OnClientSide bool
   272  	// Message can be a proto.Message or []byte. Other messages formats are not
   273  	// supported.
   274  	Message any
   275  }
   276  
   277  func (c *ServerMessage) toProto() *binlogpb.GrpcLogEntry {
   278  	var (
   279  		data []byte
   280  		err  error
   281  	)
   282  	if m, ok := c.Message.(proto.Message); ok {
   283  		data, err = proto.Marshal(m)
   284  		if err != nil {
   285  			grpclogLogger.Infof("binarylogging: failed to marshal proto message: %v", err)
   286  		}
   287  	} else if b, ok := c.Message.([]byte); ok {
   288  		data = b
   289  	} else {
   290  		grpclogLogger.Infof("binarylogging: message to log is neither proto.message nor []byte")
   291  	}
   292  	ret := &binlogpb.GrpcLogEntry{
   293  		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_MESSAGE,
   294  		Payload: &binlogpb.GrpcLogEntry_Message{
   295  			Message: &binlogpb.Message{
   296  				Length: uint32(len(data)),
   297  				Data:   data,
   298  			},
   299  		},
   300  	}
   301  	if c.OnClientSide {
   302  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
   303  	} else {
   304  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
   305  	}
   306  	return ret
   307  }
   308  
   309  // ClientHalfClose configs the binary log entry to be a ClientHalfClose entry.
   310  type ClientHalfClose struct {
   311  	OnClientSide bool
   312  }
   313  
   314  func (c *ClientHalfClose) toProto() *binlogpb.GrpcLogEntry {
   315  	ret := &binlogpb.GrpcLogEntry{
   316  		Type:    binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_HALF_CLOSE,
   317  		Payload: nil, // No payload here.
   318  	}
   319  	if c.OnClientSide {
   320  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
   321  	} else {
   322  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
   323  	}
   324  	return ret
   325  }
   326  
   327  // ServerTrailer configs the binary log entry to be a ServerTrailer entry.
   328  type ServerTrailer struct {
   329  	OnClientSide bool
   330  	Trailer      metadata.MD
   331  	// Err is the status error.
   332  	Err error
   333  	// PeerAddr is required only when it's on client side and the RPC is trailer
   334  	// only.
   335  	PeerAddr net.Addr
   336  }
   337  
   338  func (c *ServerTrailer) toProto() *binlogpb.GrpcLogEntry {
   339  	st, ok := status.FromError(c.Err)
   340  	if !ok {
   341  		grpclogLogger.Info("binarylogging: error in trailer is not a status error")
   342  	}
   343  	var (
   344  		detailsBytes []byte
   345  		err          error
   346  	)
   347  	stProto := st.Proto()
   348  	if stProto != nil && len(stProto.Details) != 0 {
   349  		detailsBytes, err = proto.Marshal(stProto)
   350  		if err != nil {
   351  			grpclogLogger.Infof("binarylogging: failed to marshal status proto: %v", err)
   352  		}
   353  	}
   354  	ret := &binlogpb.GrpcLogEntry{
   355  		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_TRAILER,
   356  		Payload: &binlogpb.GrpcLogEntry_Trailer{
   357  			Trailer: &binlogpb.Trailer{
   358  				Metadata:      mdToMetadataProto(c.Trailer),
   359  				StatusCode:    uint32(st.Code()),
   360  				StatusMessage: st.Message(),
   361  				StatusDetails: detailsBytes,
   362  			},
   363  		},
   364  	}
   365  	if c.OnClientSide {
   366  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
   367  	} else {
   368  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
   369  	}
   370  	if c.PeerAddr != nil {
   371  		ret.Peer = addrToProto(c.PeerAddr)
   372  	}
   373  	return ret
   374  }
   375  
   376  // Cancel configs the binary log entry to be a Cancel entry.
   377  type Cancel struct {
   378  	OnClientSide bool
   379  }
   380  
   381  func (c *Cancel) toProto() *binlogpb.GrpcLogEntry {
   382  	ret := &binlogpb.GrpcLogEntry{
   383  		Type:    binlogpb.GrpcLogEntry_EVENT_TYPE_CANCEL,
   384  		Payload: nil,
   385  	}
   386  	if c.OnClientSide {
   387  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
   388  	} else {
   389  		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
   390  	}
   391  	return ret
   392  }
   393  
   394  // metadataKeyOmit returns whether the metadata entry with this key should be
   395  // omitted.
   396  func metadataKeyOmit(key string) bool {
   397  	switch key {
   398  	case "lb-token", ":path", ":authority", "content-encoding", "content-type", "user-agent", "te":
   399  		return true
   400  	case "grpc-trace-bin": // grpc-trace-bin is special because it's visible to users.
   401  		return false
   402  	}
   403  	return strings.HasPrefix(key, "grpc-")
   404  }
   405  
   406  func mdToMetadataProto(md metadata.MD) *binlogpb.Metadata {
   407  	ret := &binlogpb.Metadata{}
   408  	for k, vv := range md {
   409  		if metadataKeyOmit(k) {
   410  			continue
   411  		}
   412  		for _, v := range vv {
   413  			ret.Entry = append(ret.Entry,
   414  				&binlogpb.MetadataEntry{
   415  					Key:   k,
   416  					Value: []byte(v),
   417  				},
   418  			)
   419  		}
   420  	}
   421  	return ret
   422  }
   423  
   424  func addrToProto(addr net.Addr) *binlogpb.Address {
   425  	ret := &binlogpb.Address{}
   426  	switch a := addr.(type) {
   427  	case *net.TCPAddr:
   428  		if a.IP.To4() != nil {
   429  			ret.Type = binlogpb.Address_TYPE_IPV4
   430  		} else if a.IP.To16() != nil {
   431  			ret.Type = binlogpb.Address_TYPE_IPV6
   432  		} else {
   433  			ret.Type = binlogpb.Address_TYPE_UNKNOWN
   434  			// Do not set address and port fields.
   435  			break
   436  		}
   437  		ret.Address = a.IP.String()
   438  		ret.IpPort = uint32(a.Port)
   439  	case *net.UnixAddr:
   440  		ret.Type = binlogpb.Address_TYPE_UNIX
   441  		ret.Address = a.String()
   442  	default:
   443  		ret.Type = binlogpb.Address_TYPE_UNKNOWN
   444  	}
   445  	return ret
   446  }
   447
View as plain text