mirror of
https://github.com/status-im/consul.git
synced 2025-01-09 13:26:07 +00:00
326c0ecfbe
* Implement In-Process gRPC for use by controller caching/indexing This replaces the pipe base listener implementation we were previously using. The new style CAN avoid cloning resources which our controller caching/indexing is taking advantage of to not duplicate resource objects in memory. To maintain safety for controllers and for them to be able to modify data they get back from the cache and the resource service, the client they are presented in their runtime will be wrapped with an autogenerated client which clones request and response messages as they pass through the client. Another sizable change in this PR is to consolidate how server specific gRPC services get registered and managed. Before this was in a bunch of different methods and it was difficult to track down how gRPC services were registered. Now its all in one place. * Fix race in tests * Ensure the resource service is registered to the multiplexed handler for forwarding from client agents * Expose peer streaming on the internal handler
108 lines
3.2 KiB
Go
108 lines
3.2 KiB
Go
// Copyright (c) HashiCorp, Inc.
|
|
// SPDX-License-Identifier: BUSL-1.1
|
|
|
|
package operator
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/acl/resolver"
|
|
external "github.com/hashicorp/consul/agent/grpc-external"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/proto/private/pboperator"
|
|
"github.com/hashicorp/go-hclog"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
// For private/internal gRPC handlers, protoc-gen-rpc-glue generates the
|
|
// requisite methods to satisfy the structs.RPCInfo interface using fields
|
|
// from the pbcommon package. This service is public, so we can't use those
|
|
// fields in our proto definition. Instead, we construct our RPCInfo manually.
|
|
var writeRequest struct {
|
|
structs.WriteRequest
|
|
structs.DCSpecificRequest
|
|
}
|
|
|
|
var readRequest struct {
|
|
structs.QueryOptions
|
|
structs.DCSpecificRequest
|
|
}
|
|
|
|
// Server implements pboperator.OperatorService to provide RPC operations for
|
|
// managing operator operation.
|
|
type Server struct {
|
|
Config
|
|
}
|
|
|
|
func (s *Server) TransferLeader(ctx context.Context, request *pboperator.TransferLeaderRequest) (*pboperator.TransferLeaderResponse, error) {
|
|
resp := &pboperator.TransferLeaderResponse{Success: false}
|
|
handled, err := s.ForwardRPC(&writeRequest, func(conn *grpc.ClientConn) error {
|
|
ctx := external.ForwardMetadataContext(ctx)
|
|
var err error
|
|
resp, err = pboperator.NewOperatorServiceClient(conn).TransferLeader(ctx, request)
|
|
return err
|
|
})
|
|
if handled || err != nil {
|
|
return resp, err
|
|
}
|
|
|
|
var authzCtx acl.AuthorizerContext
|
|
entMeta := structs.DefaultEnterpriseMetaInDefaultPartition()
|
|
|
|
options, err := external.QueryOptionsFromContext(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
authz, err := s.Backend.ResolveTokenAndDefaultMeta(options.Token, entMeta, &authzCtx)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
|
|
if err := authz.ToAllowAuthorizer().OperatorWriteAllowed(&authzCtx); err != nil {
|
|
return resp, err
|
|
}
|
|
|
|
return s.Backend.TransferLeader(ctx, request)
|
|
}
|
|
|
|
type Config struct {
|
|
Backend Backend
|
|
Logger hclog.Logger
|
|
ForwardRPC func(structs.RPCInfo, func(*grpc.ClientConn) error) (bool, error)
|
|
Datacenter string
|
|
}
|
|
|
|
func NewServer(cfg Config) *Server {
|
|
requireNotNil(cfg.Backend, "Backend")
|
|
requireNotNil(cfg.Logger, "Logger")
|
|
requireNotNil(cfg.ForwardRPC, "ForwardRPC")
|
|
if cfg.Datacenter == "" {
|
|
panic("Datacenter is required")
|
|
}
|
|
return &Server{
|
|
Config: cfg,
|
|
}
|
|
}
|
|
|
|
func requireNotNil(v interface{}, name string) {
|
|
if v == nil {
|
|
panic(name + " is required")
|
|
}
|
|
}
|
|
|
|
var _ pboperator.OperatorServiceServer = (*Server)(nil)
|
|
|
|
func (s *Server) Register(registrar grpc.ServiceRegistrar) {
|
|
pboperator.RegisterOperatorServiceServer(registrar, s)
|
|
}
|
|
|
|
// Backend defines the core integrations the Operator endpoint depends on. A
|
|
// functional implementation will integrate with various operator operation such as
|
|
// raft, autopilot operation. The only currently implemented operation is raft leader transfer
|
|
type Backend interface {
|
|
TransferLeader(ctx context.Context, request *pboperator.TransferLeaderRequest) (*pboperator.TransferLeaderResponse, error)
|
|
ResolveTokenAndDefaultMeta(token string, entMeta *acl.EnterpriseMeta, authzCtx *acl.AuthorizerContext) (resolver.Result, error)
|
|
}
|