service.go

Documentation: google.golang.org/grpc/profiling/service

     1  /*
     2   *
     3   * Copyright 2019 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 service defines methods to register a gRPC client/service for a
    20  // profiling service that is exposed in the same server. This service can be
    21  // queried by a client to remotely manage the gRPC profiling behaviour of an
    22  // application.
    23  //
    24  // # Experimental
    25  //
    26  // Notice: This package is EXPERIMENTAL and may be changed or removed in a
    27  // later release.
    28  package service
    29  
    30  import (
    31  	"context"
    32  	"errors"
    33  	"sync"
    34  
    35  	"google.golang.org/grpc"
    36  	"google.golang.org/grpc/grpclog"
    37  	"google.golang.org/grpc/internal/profiling"
    38  	ppb "google.golang.org/grpc/profiling/proto"
    39  )
    40  
    41  var logger = grpclog.Component("profiling")
    42  
    43  // ProfilingConfig defines configuration options for the Init method.
    44  type ProfilingConfig struct {
    45  	// Setting this to true will enable profiling.
    46  	Enabled bool
    47  
    48  	// Profiling uses a circular buffer (ring buffer) to store statistics for
    49  	// only the last few RPCs so that profiling stats do not grow unbounded. This
    50  	// parameter defines the upper limit on the number of RPCs for which
    51  	// statistics should be stored at any given time. An average RPC requires
    52  	// approximately 2-3 KiB of memory for profiling-related statistics, so
    53  	// choose an appropriate number based on the amount of memory you can afford.
    54  	StreamStatsSize uint32
    55  
    56  	// To expose the profiling service and its methods, a *grpc.Server must be
    57  	// provided.
    58  	Server *grpc.Server
    59  }
    60  
    61  var errorNilServer = errors.New("profiling: no grpc.Server provided")
    62  
    63  // Init takes a *ProfilingConfig to initialize profiling (turned on/off
    64  // depending on the value set in pc.Enabled) and register the profiling service
    65  // in the server provided in pc.Server.
    66  func Init(pc *ProfilingConfig) error {
    67  	if pc.Server == nil {
    68  		return errorNilServer
    69  	}
    70  
    71  	if err := profiling.InitStats(pc.StreamStatsSize); err != nil {
    72  		return err
    73  	}
    74  
    75  	ppb.RegisterProfilingServer(pc.Server, getProfilingServerInstance())
    76  
    77  	// Do this last after everything has been initialized and allocated.
    78  	profiling.Enable(pc.Enabled)
    79  
    80  	return nil
    81  }
    82  
    83  type profilingServer struct {
    84  	ppb.UnimplementedProfilingServer
    85  	drainMutex sync.Mutex
    86  }
    87  
    88  var profilingServerInstance *profilingServer
    89  var profilingServerOnce sync.Once
    90  
    91  // getProfilingServerInstance creates and returns a singleton instance of
    92  // profilingServer. Only one instance of profilingServer is created to use a
    93  // shared mutex across all profilingServer instances.
    94  func getProfilingServerInstance() *profilingServer {
    95  	profilingServerOnce.Do(func() {
    96  		profilingServerInstance = &profilingServer{}
    97  	})
    98  
    99  	return profilingServerInstance
   100  }
   101  
   102  func (s *profilingServer) Enable(ctx context.Context, req *ppb.EnableRequest) (*ppb.EnableResponse, error) {
   103  	if req.Enabled {
   104  		logger.Infof("profilingServer: Enable: enabling profiling")
   105  	} else {
   106  		logger.Infof("profilingServer: Enable: disabling profiling")
   107  	}
   108  	profiling.Enable(req.Enabled)
   109  
   110  	return &ppb.EnableResponse{}, nil
   111  }
   112  
   113  func timerToProtoTimer(timer *profiling.Timer) *ppb.Timer {
   114  	return &ppb.Timer{
   115  		Tags:      timer.Tags,
   116  		BeginSec:  timer.Begin.Unix(),
   117  		BeginNsec: int32(timer.Begin.Nanosecond()),
   118  		EndSec:    timer.End.Unix(),
   119  		EndNsec:   int32(timer.End.Nanosecond()),
   120  		GoId:      timer.GoID,
   121  	}
   122  }
   123  
   124  func statToProtoStat(stat *profiling.Stat) *ppb.Stat {
   125  	protoStat := &ppb.Stat{
   126  		Tags:     stat.Tags,
   127  		Timers:   make([]*ppb.Timer, 0, len(stat.Timers)),
   128  		Metadata: stat.Metadata,
   129  	}
   130  	for _, t := range stat.Timers {
   131  		protoStat.Timers = append(protoStat.Timers, timerToProtoTimer(t))
   132  	}
   133  	return protoStat
   134  }
   135  
   136  func (s *profilingServer) GetStreamStats(ctx context.Context, req *ppb.GetStreamStatsRequest) (*ppb.GetStreamStatsResponse, error) {
   137  	// Since the drain operation is destructive, only one client request should
   138  	// be served at a time.
   139  	logger.Infof("profilingServer: GetStreamStats: processing request")
   140  	s.drainMutex.Lock()
   141  	results := profiling.StreamStats.Drain()
   142  	s.drainMutex.Unlock()
   143  
   144  	logger.Infof("profilingServer: GetStreamStats: returning %v records", len(results))
   145  	streamStats := make([]*ppb.Stat, 0)
   146  	for _, stat := range results {
   147  		streamStats = append(streamStats, statToProtoStat(stat.(*profiling.Stat)))
   148  	}
   149  	return &ppb.GetStreamStatsResponse{StreamStats: streamStats}, nil
   150  }
   151
View as plain text