diff options
author | Pavlo Strokov <pstrokov@gitlab.com> | 2020-06-26 14:16:27 +0300 |
---|---|---|
committer | Pavlo Strokov <pstrokov@gitlab.com> | 2020-06-26 14:16:27 +0300 |
commit | 389ae3336772c9bbf7cd21f2f6aed262e8ae0357 (patch) | |
tree | ebcbff78afafb87cc341784c9839388308f68565 /internal/praefect/server_factory.go | |
parent | e57acd088cf151b6cecc70b5c7812955b3f49c93 (diff) |
Praefect: server factory introduction
Introduces server factory for creating gRPC servers.
Praefect gRPC server created by separate function and
can be reused in tests to check routing.
Part of: https://gitlab.com/gitlab-org/gitaly/-/issues/1698
Diffstat (limited to 'internal/praefect/server_factory.go')
-rw-r--r-- | internal/praefect/server_factory.go | 117 |
1 files changed, 117 insertions, 0 deletions
diff --git a/internal/praefect/server_factory.go b/internal/praefect/server_factory.go new file mode 100644 index 000000000..90309380e --- /dev/null +++ b/internal/praefect/server_factory.go @@ -0,0 +1,117 @@ +package praefect + +import ( + "net" + "sync" + + "github.com/sirupsen/logrus" + "gitlab.com/gitlab-org/gitaly/internal/praefect/config" + "gitlab.com/gitlab-org/gitaly/internal/praefect/datastore" + "gitlab.com/gitlab-org/gitaly/internal/praefect/grpc-proxy/proxy" + "gitlab.com/gitlab-org/gitaly/internal/praefect/nodes" + "gitlab.com/gitlab-org/gitaly/internal/praefect/protoregistry" + "gitlab.com/gitlab-org/gitaly/internal/praefect/transactions" + "google.golang.org/grpc" +) + +// NewServerFactory returns factory object for initialization of praefect gRPC servers. +func NewServerFactory( + conf config.Config, + logger *logrus.Entry, + director proxy.StreamDirector, + nodeMgr nodes.Manager, + txMgr *transactions.Manager, + queue datastore.ReplicationEventQueue, + registry *protoregistry.Registry, +) *ServerFactory { + return &ServerFactory{ + conf: conf, + logger: logger, + director: director, + nodeMgr: nodeMgr, + txMgr: txMgr, + queue: queue, + registry: registry, + } +} + +// ServerFactory is a factory of praefect grpc servers +type ServerFactory struct { + mtx sync.Mutex + conf config.Config + logger *logrus.Entry + director proxy.StreamDirector + nodeMgr nodes.Manager + txMgr *transactions.Manager + queue datastore.ReplicationEventQueue + registry *protoregistry.Registry + insecure []*grpc.Server +} + +// Serve starts serving on the provided listener with newly created grpc.Server +func (s *ServerFactory) Serve(l net.Listener, secure bool) error { + srv, err := s.create() + if err != nil { + return err + } + + return srv.Serve(l) +} + +// Stop stops all servers created by the factory. +func (s *ServerFactory) Stop() { + for _, srv := range s.all() { + srv.Stop() + } +} + +// GracefulStop stops both the secure and insecure servers gracefully. +func (s *ServerFactory) GracefulStop() { + wg := sync.WaitGroup{} + + for _, srv := range s.all() { + wg.Add(1) + + go func(s *grpc.Server) { + s.GracefulStop() + wg.Done() + }(srv) + } + + wg.Wait() +} + +func (s *ServerFactory) create() (*grpc.Server, error) { + s.mtx.Lock() + defer s.mtx.Unlock() + + s.insecure = append(s.insecure, s.createGRPC()) + + return s.insecure[len(s.insecure)-1], nil +} + +func (s *ServerFactory) createGRPC(grpcOpts ...grpc.ServerOption) *grpc.Server { + return NewGRPCServer( + s.conf, + s.logger, + s.registry, + s.director, + s.nodeMgr, + s.txMgr, + s.queue, + grpcOpts..., + ) +} + +func (s *ServerFactory) all() []*grpc.Server { + s.mtx.Lock() + defer s.mtx.Unlock() + + var servers []*grpc.Server + + if s.insecure != nil { + servers = append(servers, s.insecure...) + } + + return servers +} |