mnemosyned/daemon.go
func (d *Daemon) Run() (err error) {
	var (
		cl     *cluster.Cluster
		tracer opentracing.Tracer
	)
	if cl, err = initCluster(d.logger, d.opts.ClusterListenAddr, d.opts.ClusterSeeds...); err != nil {
		return
	}
	if err = d.initStorage(d.logger, d.opts.PostgresTable, d.opts.PostgresSchema); err != nil {
		return
	}

	interceptor := promgrpc.NewInterceptor(promgrpc.InterceptorOpts{})

	if d.opts.TracingAgentAddress != "" {
		if tracer, d.tracerCloser, err = initJaeger(
			constant.Subsystem,
			d.opts.ClusterListenAddr,
			d.opts.TracingAgentAddress,
			d.logger.Named("tracer"),
		); err != nil {
			return
		}
		d.logger.Info("tracing enabled", zap.String("agent_address", d.opts.TracingAgentAddress))
	} else {
		tracer = opentracing.NoopTracer{}
	}

	d.clientOptions = []grpc.DialOption{
		// User agent is required for example to determine if incoming request is internal.
		grpc.WithUserAgent(fmt.Sprintf("%s:%s", constant.Subsystem, d.opts.Version)),
		grpc.WithStatsHandler(interceptor),
		grpc.WithDialer(interceptor.Dialer(func(addr string, timeout time.Duration) (net.Conn, error) {
			return net.DialTimeout("tcp", addr, timeout)
		})),
		grpc.WithUnaryInterceptor(unaryClientInterceptors(
			interceptor.UnaryClient(),
			otgrpc.OpenTracingClientInterceptor(tracer)),
		),
		grpc.WithStreamInterceptor(interceptor.StreamClient()),
	}
	d.serverOptions = []grpc.ServerOption{
		grpc.StatsHandler(interceptor),
		grpc.UnaryInterceptor(unaryServerInterceptors(
			otgrpc.OpenTracingServerInterceptor(tracer),
			errorInterceptor(d.logger),
			interceptor.UnaryServer(),
		)),
	}
	if d.opts.TLS {
		servCreds, err := credentials.NewServerTLSFromFile(d.opts.TLSCertFile, d.opts.TLSKeyFile)
		if err != nil {
			return err
		}
		d.serverOptions = append(d.serverOptions, grpc.Creds(servCreds))

		clientCreds, err := credentials.NewClientTLSFromFile(d.opts.TLSCertFile, "")
		if err != nil {
			return err
		}
		d.clientOptions = append(d.clientOptions, grpc.WithTransportCredentials(clientCreds))
	} else {
		d.clientOptions = append(d.clientOptions, grpc.WithInsecure())
	}

	d.server = grpc.NewServer(d.serverOptions...)

	cache := cache.New(5*time.Second, constant.Subsystem)
	mnemosyneServer, err := newSessionManager(sessionManagerOpts{
		addr:    d.opts.ClusterListenAddr,
		cluster: cl,
		logger:  d.logger,
		storage: d.storage,
		ttc:     d.opts.SessionTTC,
		cache:   cache,
		tracer:  tracer,
	})
	if err != nil {
		return err
	}

	mnemosynerpc.RegisterSessionManagerServer(d.server, mnemosyneServer)
	grpc_health_v1.RegisterHealthServer(d.server, health.NewServer())

	if !d.opts.IsTest {
		prometheus.DefaultRegisterer.Register(d.storage.(storage.InstrumentedStorage))
		prometheus.DefaultRegisterer.Register(cache)
		prometheus.DefaultRegisterer.Register(mnemosyneServer)
		prometheus.DefaultRegisterer.Register(interceptor)
		promgrpc.RegisterInterceptor(d.server, interceptor)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err = cl.Connect(ctx, d.clientOptions...); err != nil {
		return err
	}

	go func() {
		d.logger.Info("rpc server is running", zap.String("address", d.rpcListener.Addr().String()))

		if err := d.server.Serve(d.rpcListener); err != nil {
			if err == grpc.ErrServerStopped {
				d.logger.Info("grpc server has been stopped")
				return
			}

			if !strings.Contains(err.Error(), "use of closed network connection") {
				d.logger.Error("rpc server failure", zap.Error(err))
			}
		}
	}()

	if d.debugListener != nil {
		go func() {
			d.logger.Info("debug server is running", zap.String("address", d.debugListener.Addr().String()))

			mux := http.NewServeMux()
			mux.Handle("/debug/pprof/", http.HandlerFunc(pprof.Index))
			mux.Handle("/debug/pprof/cmdline", http.HandlerFunc(pprof.Cmdline))
			mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
			mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
			mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
			mux.Handle("/metrics", promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{}))
			mux.Handle("/healthz", &livenessHandler{
				livenessResponse: livenessResponse{
					Version: d.opts.Version,
				},
				logger: d.logger,
			})
			mux.Handle("/healthr", &readinessHandler{
				livenessResponse: livenessResponse{
					Version: d.opts.Version,
				},
				logger:   d.logger,
				postgres: d.postgres,
				cluster:  cl,
			})
			if err := http.Serve(d.debugListener, mux); err != nil {
				d.logger.Error("debug server failure", zap.Error(err))
			}
		}()
	}

	go mnemosyneServer.cleanup(d.done)

	return
}
func (d *Daemon) Run() (err error) {
	var (
		cl     *cluster.Cluster
		tracer opentracing.Tracer
	)
	if cl, err = initCluster(d.logger, d.opts.ClusterListenAddr, d.opts.ClusterSeeds...); err != nil {
		return
	}
	if err = d.initStorage(d.logger, d.opts.PostgresTable, d.opts.PostgresSchema); err != nil {
		return
	}

	interceptor := promgrpc.NewInterceptor(promgrpc.InterceptorOpts{})

	if d.opts.TracingAgentAddress != "" {
		if tracer, d.tracerCloser, err = initJaeger(
			constant.Subsystem,
			d.opts.ClusterListenAddr,
			d.opts.TracingAgentAddress,
			d.logger.Named("tracer"),
		); err != nil {
			return
		}
		d.logger.Info("tracing enabled", zap.String("agent_address", d.opts.TracingAgentAddress))
	} else {
		tracer = opentracing.NoopTracer{}
	}

	d.clientOptions = []grpc.DialOption{
		// User agent is required for example to determine if incoming request is internal.
		grpc.WithUserAgent(fmt.Sprintf("%s:%s", constant.Subsystem, d.opts.Version)),
		grpc.WithStatsHandler(interceptor),
		grpc.WithDialer(interceptor.Dialer(func(addr string, timeout time.Duration) (net.Conn, error) {
			return net.DialTimeout("tcp", addr, timeout)
		})),
		grpc.WithUnaryInterceptor(unaryClientInterceptors(
			interceptor.UnaryClient(),
			otgrpc.OpenTracingClientInterceptor(tracer)),
		),
		grpc.WithStreamInterceptor(interceptor.StreamClient()),
	}
	d.serverOptions = []grpc.ServerOption{
		grpc.StatsHandler(interceptor),
		grpc.UnaryInterceptor(unaryServerInterceptors(
			otgrpc.OpenTracingServerInterceptor(tracer),
			errorInterceptor(d.logger),
			interceptor.UnaryServer(),
		)),
	}
	if d.opts.TLS {
		servCreds, err := credentials.NewServerTLSFromFile(d.opts.TLSCertFile, d.opts.TLSKeyFile)
		if err != nil {
			return err
		}
		d.serverOptions = append(d.serverOptions, grpc.Creds(servCreds))

		clientCreds, err := credentials.NewClientTLSFromFile(d.opts.TLSCertFile, "")
		if err != nil {
			return err
		}
		d.clientOptions = append(d.clientOptions, grpc.WithTransportCredentials(clientCreds))
	} else {
		d.clientOptions = append(d.clientOptions, grpc.WithInsecure())
	}

	d.server = grpc.NewServer(d.serverOptions...)

	cache := cache.New(5*time.Second, constant.Subsystem)
	mnemosyneServer, err := newSessionManager(sessionManagerOpts{
		addr:    d.opts.ClusterListenAddr,
		cluster: cl,
		logger:  d.logger,
		storage: d.storage,
		ttc:     d.opts.SessionTTC,
		cache:   cache,
		tracer:  tracer,
	})
	if err != nil {
		return err
	}

	mnemosynerpc.RegisterSessionManagerServer(d.server, mnemosyneServer)
	grpc_health_v1.RegisterHealthServer(d.server, health.NewServer())

	if !d.opts.IsTest {
		prometheus.DefaultRegisterer.Register(d.storage.(storage.InstrumentedStorage))
		prometheus.DefaultRegisterer.Register(cache)
		prometheus.DefaultRegisterer.Register(mnemosyneServer)
		prometheus.DefaultRegisterer.Register(interceptor)
		promgrpc.RegisterInterceptor(d.server, interceptor)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err = cl.Connect(ctx, d.clientOptions...); err != nil {
		return err
	}

	go func() {
		d.logger.Info("rpc server is running", zap.String("address", d.rpcListener.Addr().String()))

		if err := d.server.Serve(d.rpcListener); err != nil {
			if err == grpc.ErrServerStopped {
				d.logger.Info("grpc server has been stopped")
				return
			}

			if !strings.Contains(err.Error(), "use of closed network connection") {
				d.logger.Error("rpc server failure", zap.Error(err))
			}
		}
	}()

	if d.debugListener != nil {
		go func() {
			d.logger.Info("debug server is running", zap.String("address", d.debugListener.Addr().String()))

			mux := http.NewServeMux()
			mux.Handle("/debug/pprof/", http.HandlerFunc(pprof.Index))
			mux.Handle("/debug/pprof/cmdline", http.HandlerFunc(pprof.Cmdline))
			mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
			mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
			mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
			mux.Handle("/metrics", promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{}))
			mux.Handle("/healthz", &livenessHandler{
				livenessResponse: livenessResponse{
					Version: d.opts.Version,
				},
				logger: d.logger,
			})
			mux.Handle("/healthr", &readinessHandler{
				livenessResponse: livenessResponse{
					Version: d.opts.Version,
				},
				logger:   d.logger,
				postgres: d.postgres,
				cluster:  cl,
			})
			if err := http.Serve(d.debugListener, mux); err != nil {
				d.logger.Error("debug server failure", zap.Error(err))
			}
		}()
	}

	go mnemosyneServer.cleanup(d.done)

	return
}
// Run starts daemon and all services within.