gRPCがどんな動きをするのかhelloworld試しても分からなかったのでコードを読解してみた

Why I thouth it

• 業務の一環でAPIを構築することになったが、アーキテクトから組むのは初体験
• しかもざっくりとしかRESTも理解してなかった
• なので、再度勉強し直す意味も踏まえて昨今の時勢を調べてみた
• 「REST」に加え、昨今は「gRPC」などが一般的な代替手段となっているとのこと
• gRPCについては全くわからなかったので、調べながらとりあえずhelloworldを試してみた
• が、やってみてもgRPCがどんなことしてるのかサッパリわからない状態だったので、コード読んで見ることにした

[参考文献]
https://docs.microsoft.com/ja-jp/azure/architecture/microservices/api-design

Whats' gRPC?

• Googleが開発したOSSのRPC framework
• 「ロードバランシング / トレーシング / ヘルスチェック / 認証」のためのプラグができるサポートがあるため
• データセンターを跨いでの効率的なサービスインができる
• とのこと (※1 参照)

RESTとRPCの差異の一部を↓にまとめた
文章は参考文献(※2, ※3, ※4, ※5)から引用

名称	REST	RPC
正式名称	Representational State Transfer	remote procedure call
ざっくり概要	SOAPプロトコルのようなMEPベースの特別な抽象化をしないもののことを、大まかに意味する用語	PGから別のアドレス空間(同一NWの別host)にあるPGを実行することを可能にする技術
対象	HTTP 動詞に基づいて統一インターフェイスを定義	操作やコマンドを中心にインターフェイス化
特徴	HTTPリクエストの１行目を見ればクライアントの目的を基本的に理解することができる(残りはその詳細に過ぎない)	遠隔相互作用の詳細を明示的にコーディングする必要がない
アプローチ1	スコープ情報の様々な値に対して異なるURIを返す	「ドキュメントプロセッサ」ごとにURIを公開する
アプローチ2	リソース（名詞）の多様性を重視	プロトコル操作（動詞）の多様性を重視

[参考]
※1 https://grpc.io/about/
※2 https://docs.microsoft.com/ja-jp/azure/architecture/microservices/api-design
※3 https://restful-api-guidelines-ja.netlify.com
※4 https://ja.wikipedia.org/wiki/Representational_State_Transfer
※5 RESTful　Webサービス [ レオナルド・リチャードソン ]

How to gRPC's "HelloWorld"?

以下の作業が必要となるが、ここではさっくりとしか紹介しないので
詳細は是非参考にさせていただいたサイトを見ていただきたい

• gRPCを利用するための環境準備

  -- gRPCはprotocコマンドを使ってコンパイルするので↓が必要
  $ brew install protobuf
  
  -- golangで使うgRPCライブラリをDL
  $ go get -u google.golang.org/grpc
  $ go get -u github.com/golang/protobuf/protoc-gen-go
  
  -- PATHを通す（通さないとprotocでコンパイルするときにエラーになる）
  $ export PATH=$PATH:$(go env GOPATH)/bin
  $ export PATH=$PATH:$GOPATH/bin
  

• サンプルコードの準備

  -- helloworldのコードはインストールしたgrpcのディレクトリに存在する
  $ ls $GOPATH/src/google.golang.org/grpc/examples/helloworld/
  $ tree $GOPATH/src/google.golang.org/grpc/examples/helloworld/
      .
      ├── greeter_client
      │   └── main.go
      ├── greeter_server
      │   └── main.go
      ├── helloworld
      │   ├── helloworld.pb.go
      │   └── helloworld.proto
      └── mock_helloworld
          ├── hw_mock.go
          └── hw_mock_test.go
  
  -- 作業用にcpする
  $ cp -r $GOPATH/src/google.golang.org/grpc/examples/helloworld .
  $ cd helloworld
  

• 動かして動作確認

  -- Server側を起動する
  $ go run greeter_server/main.go
  
  -- 別タブでClient側を実行して動作検証
  $ go run greeter_client/main.go
  

[参考]
https://budougumi0617.github.io/2018/01/01/hello-grpc-go/

helloworldがどうやって動いているかを読み解いていく

↓のGitHubのコードやリンクを見ながら紐解いていく
https://github.com/grpc/grpc-go/tree/master/examples/helloworld

< Side greeter_server >

■ greeter_server/main.go でやっていることを理解する

greeter_server/main.go

func main() {
    lis, err := net.Listen("tcp", port)
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    s := grpc.NewServer()
    pb.RegisterGreeterServer(s, &server{})
    // Register reflection service on gRPC server.
    reflection.Register(s)
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

→ 上から順に紐解いていく。

lis, err := net.Listen("tcp", port)

引数に渡している 「port」は「:50051」が入っている↓

    const (
        port = ":50051"
    )

• このメソッドを定義してるのはnetパッケージ = netライブラリのメソッド
• 処理の内容としては、「localhost:50051をListen状態にしている」とのこと
• 返り値に Listener struct を返す模様

s := grpc.NewServer()

このメソッドの定義があったのは grpc-go/server.go
↓詳細

// NewServer creates a gRPC server which has no service registered and has not
// started to accept requests yet.
func NewServer(opt ...ServerOption) *Server {
    opts := defaultServerOptions
    for _, o := range opt {
        o(&opts)
    }
    s := &Server{
        lis:    make(map[net.Listener]bool),
        opts:   opts,
        conns:  make(map[io.Closer]bool),
        m:      make(map[string]*service),
        quit:   make(chan struct{}),
        done:   make(chan struct{}),
        czData: new(channelzData),
    }
    s.cv = sync.NewCond(&s.mu)
    if EnableTracing {
        _, file, line, _ := runtime.Caller(1)
        s.events = trace.NewEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
    }

    if channelz.IsOn() {
        s.channelzID = channelz.RegisterServer(&channelzServer{s}, "")
    }
    return s
}

• 「opts := defaultServerOptions」 =　ServerOption options structs の　一部フィールドに 値を代入

  var defaultServerOptions = options{
      maxReceiveMessageSize: defaultServerMaxReceiveMessageSize,
      maxSendMessageSize:    defaultServerMaxSendMessageSize,
      connectionTimeout:     120 * time.Second,
      writeBufferSize:       defaultWriteBufSize,
      readBufferSize:        defaultReadBufSize,
  }
  

• （あれば）ServerOption型( options structs )の引数で受け取った設定を、opts(defaultServerOptions)に代入する

  o(&opts)
  
  --------
  
  // A ServerOption sets options such as credentials, codec and keepalive parameters, etc.
  type ServerOption func(*options)
  
  --------
  
  type options struct {
      creds                 credentials.TransportCredentials
      codec                 baseCodec
      cp                    Compressor
      dc                    Decompressor
      unaryInt              UnaryServerInterceptor
      streamInt             StreamServerInterceptor
      inTapHandle           tap.ServerInHandle
      statsHandler          stats.Handler
      maxConcurrentStreams  uint32
      maxReceiveMessageSize int
      maxSendMessageSize    int
      unknownStreamDesc     *StreamDesc
      keepaliveParams       keepalive.ServerParameters
      keepalivePolicy       keepalive.EnforcementPolicy
      initialWindowSize     int32
      initialConnWindowSize int32
      writeBufferSize       int
      readBufferSize        int
      connectionTimeout     time.Duration
      maxHeaderListSize     *uint32
  }
  

• Server structs に各種値を入れる

  // Server is a gRPC server to serve RPC requests.
  type Server struct {
      opts options
  
      mu     sync.Mutex // guards following
      lis    map[net.Listener]bool
      conns  map[io.Closer]bool
      serve  bool
      drain  bool
      cv     *sync.Cond          // signaled when connections close for GracefulStop
      m      map[string]*service // service name -> service info
      events trace.EventLog
  
      quit               chan struct{}
      done               chan struct{}
      quitOnce           sync.Once
      doneOnce           sync.Once
      channelzRemoveOnce sync.Once
      serveWG            sync.WaitGroup // counts active Serve goroutines for GracefulStop
  
      channelzID int64 // channelz unique identification number
      czData     *channelzData
  }
  

• 「s.cv = sync.NewCond(&s.mu)」 = r/wのlock管理を設定する

  • NewCond = read/writeのlockとunlockを管理する 用のメソッドらしい

    func NewCond(l Locker) *Cond
    
    -----
    
    type Locker interface {
            Lock()
            Unlock()
    }
    
    -----
    
    > func (rw *RWMutex) RLocker() Locker
    RLocker returns a Locker interface that implements the Lock and Unlock methods by calling rw.RLock and rw.RUnlock.
    

• RPCsが利用できる状態かどうかをチェック

• channelz(gRPCの中で異なったレベルで接続しているランタイム情報を提供するツール) がonになっているか確認

• Server structsを返す

pb.RegisterGreeterServer(s, &server{})

「pb = helloworld」と定義しているので、helloworldディレクトリで定義した RegisterGreeterServer() である

func RegisterGreeterServer(s *grpc.Server, srv GreeterServer) {
    s.RegisterService(&_Greeter_serviceDesc, srv)
}

↑で使われている RegisterService()の詳細↓

func (s *Server) RegisterService(sd *ServiceDesc, ss interface{}) {
    ht := reflect.TypeOf(sd.HandlerType).Elem()
    st := reflect.TypeOf(ss)
    if !st.Implements(ht) {
        grpclog.Fatalf("grpc: Server.RegisterService found the handler of type %v that does not satisfy %v", st, ht)
    }
    s.register(sd, ss)
}

→ 引数として渡されてきた &_Greeter_serviceDesc の型チェック & Serverに helloworldのMethodsを登録

var _Greeter_serviceDesc = grpc.ServiceDesc{
    ServiceName: "helloworld.Greeter",
    HandlerType: (*GreeterServer)(nil),
    Methods: []grpc.MethodDesc{
        {
            MethodName: "SayHello",
            Handler:    _Greeter_SayHello_Handler,
        },
    },
    Streams:  []grpc.StreamDesc{},
    Metadata: "helloworld.proto",
}

------------------------------------------------------------------------------------------------------------------

func (s *Server) register(sd *ServiceDesc, ss interface{}) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.printf("RegisterService(%q)", sd.ServiceName)
    if s.serve {
        grpclog.Fatalf("grpc: Server.RegisterService after Server.Serve for %q", sd.ServiceName)
    }
    if _, ok := s.m[sd.ServiceName]; ok {
        grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
    }
    srv := &service{
        server: ss,
        md:     make(map[string]*MethodDesc),
        sd:     make(map[string]*StreamDesc),
        mdata:  sd.Metadata,
    }
    for i := range sd.Methods {
        d := &sd.Methods[i]
        srv.md[d.MethodName] = d
    }
    for i := range sd.Streams {
        d := &sd.Streams[i]
        srv.sd[d.StreamName] = d
    }
    s.m[sd.ServiceName] = srv
}

reflection.Register(s)

grpc-go/reflectionの serverreflection.go に定義ある

// Register registers the server reflection service on the given gRPC server.
func Register(s *grpc.Server) {
    rpb.RegisterServerReflectionServer(s, &serverReflectionServer{
        s: s,
    })
}

→　rpb = "google.golang.org/grpc/reflection/grpc_reflection_v1alpha" に処理を中継

func RegisterServerReflectionServer(s *grpc.Server, srv ServerReflectionServer) {
    s.RegisterService(&_ServerReflection_serviceDesc, srv)
}

→　_ServerReflection_serviceDesc をServerに登録

var _ServerReflection_serviceDesc = grpc.ServiceDesc{
    ServiceName: "grpc.reflection.v1alpha.ServerReflection",
    HandlerType: (*ServerReflectionServer)(nil),
    Methods:     []grpc.MethodDesc{},
    Streams: []grpc.StreamDesc{
        {
            StreamName:    "ServerReflectionInfo",
            Handler:       _ServerReflection_ServerReflectionInfo_Handler,
            ServerStreams: true,
            ClientStreams: true,
        },
    },
    Metadata: "grpc_reflection_v1alpha/reflection.proto",
}

< Side greeter_client >

■ greeter_client/main.go でやっていることを理解する

func main() {
    // Set up a connection to the server.
    conn, err := grpc.Dial(address, grpc.WithInsecure())
    if err != nil {
        log.Fatalf("did not connect: %v", err)
    }
    defer conn.Close()
    c := pb.NewGreeterClient(conn)


    // Contact the server and print out its response.
    name := defaultName
    if len(os.Args) > 1 {
        name = os.Args[1]
    }
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    r, err := c.SayHello(ctx, &pb.HelloRequest{Name: name})
    if err != nil {
        log.Fatalf("could not greet: %v", err)
    }
    log.Printf("Greeting: %s", r.Message)
}

conn, err := grpc.Dial(address, grpc.WithInsecure())

client系処理は grpc-go/clientconn.go に記載されている

// Dial creates a client connection to the given target.
func Dial(target string, opts ...DialOption) (*ClientConn, error) {
    return DialContext(context.Background(), target, opts...)
}

→　DialContext という他に比べると長いメソッドを返している
　　この処理の詳細はコメントに丁寧にまとまっているので、理解するのはここを見るのが一番早かった

// DialContext creates a client connection to the given target. By default, it's
// a non-blocking dial (the function won't wait for connections to be
// established, and connecting happens in the background). To make it a blocking
// dial, use WithBlock() dial option.
//
// In the non-blocking case, the ctx does not act against the connection. It
// only controls the setup steps.
//
// In the blocking case, ctx can be used to cancel or expire the pending
// connection. Once this function returns, the cancellation and expiration of
// ctx will be noop. Users should call ClientConn.Close to terminate all the
// pending operations after this function returns.
//
// The target name syntax is defined in
// https://github.com/grpc/grpc/blob/master/doc/naming.md.
// e.g. to use dns resolver, a "dns:///" prefix should be applied to the target.

↑の処理でやったことを要約すると...
- まず渡してるものをおさらい
- context.Background() = nil以外の空のContextを返す
- target = address = :50051
- opts ...DialOption = grpc.WithInsecure() = newFuncDialOption() = funcDialOption structs = dialOptions structs へ代入した変数

    func WithInsecure() DialOption {
        return newFuncDialOption(func(o *dialOptions) {
            o.insecure = true
        })
    }

    ----------------------------------------------------

    func newFuncDialOption(f func(*dialOptions)) *funcDialOption {
        return &funcDialOption{
            f: f,
        }
    }

    ----------------------------------------------------

    // funcDialOption wraps a function that modifies dialOptions into an
    // implementation of the DialOption interface.
    type funcDialOption struct {
        f func(*dialOptions)
    }

    ----------------------------------------------------

    // dialOptions configure a Dial call. dialOptions are set by the DialOption
    // values passed to Dial.
    type dialOptions struct {
        unaryInt    UnaryClientInterceptor
        streamInt   StreamClientInterceptor
        cp          Compressor
        dc          Decompressor
        bs          backoff.Strategy
        block       bool
        insecure    bool
        timeout     time.Duration
        scChan      <-chan ServiceConfig
        authority   string
        copts       transport.ConnectOptions
        callOptions []CallOption
        // This is used by v1 balancer dial option WithBalancer to support v1
        // balancer, and also by WithBalancerName dial option.
        balancerBuilder balancer.Builder
        // This is to support grpclb.
        resolverBuilder      resolver.Builder
        reqHandshake         envconfig.RequireHandshakeSetting
        channelzParentID     int64
        disableServiceConfig bool
        disableRetry         bool
        disableHealthCheck   bool
        healthCheckFunc      internal.HealthChecker
    }

• 指定したtargetでclient接続を作成する
• デフォルトはブロックされてないからのcontext

c := pb.NewGreeterClient(conn)

server側と同じく、helloworldディレクトリで作成した NewGreeterClient()

func NewGreeterClient(cc *grpc.ClientConn) GreeterClient {
    return &greeterClient{cc}
}

→　GreeterClient interface に grpc.Dial(address, grpc.WithInsecure()) で作成した context を代入している

// GreeterClient is the client API for Greeter service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
type GreeterClient interface {
    // Sends a greeting
    SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error)
}

→　代入したContextにSayHello()を返す

残りの処理

    // Contact the server and print out its response.
    name := defaultName
    if len(os.Args) > 1 {
        name = os.Args[1]
    }
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    r, err := c.SayHello(ctx, &pb.HelloRequest{Name: name})
    if err != nil {
        log.Fatalf("could not greet: %v", err)
    }
    log.Printf("Greeting: %s", r.Message)

• defaultName は greeter_client/main.go で記述されている
• 引数が合った場合、nameを上書き
• contextのタイムアウト時間を設定する
• clientに登録したcontextに紐づけたSayHello()の実行
• 結果を標準出力

まとめ

■ protoファイル でやっていることまとめ

• 「server　ー　client」で呼び出すメソッドの定義

■ greeter_server/main.go でやっていることまとめ

• 受付portの登録
• serverとしての処理を起動するために設定を作成
• 起動させるserver定義にprotoで定義したメソッドの実体を登録
• gRPC serverとして↑で定義したserverを登録
• ↑起動

■ greeter_client/main.go でやっていることまとめ

• 空のcontextの作成
• portの登録
• 指定したtargetでclient接続を作成する
• 作成したclient接続にprotoで定義したメソッドを登録
• contextのタイムアウト時間を設定する
• clientに登録したcontextに紐づけたSayHello()の実行
• 結果を標準出力