func (s *Server) reportServer() {
	dbs := s.ClusterMetaClient.Databases()
	numDatabases := len(dbs)
	var (
		numMeasurements int64
		numSeries       int64
	)
	for _, db := range dbs {
		name := db.Name
		n, err := s.TSDBStore.SeriesCardinality(name)
		if err != nil {
			s.Logger.Error(fmt.Sprintf("Unable to get series cardinality for database %s: %v", name, err))
		} else {
			numSeries += n
		}
		n, err = s.TSDBStore.MeasurementsCardinality(name)
		if err != nil {
			s.Logger.Error(fmt.Sprintf("Unable to get measurement cardinality for database %s: %v", name, err))
		} else {
			numMeasurements += n
		}
	}
	clusterID := s.ClusterMetaClient.ClusterID()
	cl := client.New("")
	usage := client.Usage{
		Product: "influxdb",
		Data: []client.UsageData{
			{
				Values: client.Values{
					"os":               runtime.GOOS,
					"arch":             runtime.GOARCH,
					"version":          s.buildInfo.Version,
					"cluster_id":       fmt.Sprintf("%v", clusterID),
					"num_series":       numSeries,
					"num_measurements": numMeasurements,
					"num_databases":    numDatabases,
					"uptime":           time.Since(startTime).Seconds(),
				},
			},
		},
	}
	s.Logger.Info("Sending usage statistics to usage.influxdata.com")
	go cl.Save(usage)
}
reportServer reports usage statistics about the system.
package run
import (
	"crypto/tls"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"path/filepath"
	"runtime"
	"runtime/pprof"
	"time"
	"github.com/influxdata/influxdb"
	influxdb_coordinator "github.com/influxdata/influxdb/coordinator"
	"github.com/influxdata/influxdb/flux/control"
	"github.com/influxdata/influxdb/logger"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/monitor"
	"github.com/influxdata/influxdb/query"
	"github.com/influxdata/influxdb/services/collectd"
	"github.com/influxdata/influxdb/services/continuous_querier"
	"github.com/influxdata/influxdb/services/graphite"
	"github.com/influxdata/influxdb/services/httpd"
	"github.com/influxdata/influxdb/services/opentsdb"
	"github.com/influxdata/influxdb/services/precreator"
	"github.com/influxdata/influxdb/services/retention"
	"github.com/influxdata/influxdb/services/snapshotter"
	"github.com/influxdata/influxdb/services/storage"
	"github.com/influxdata/influxdb/services/subscriber"
	"github.com/influxdata/influxdb/services/udp"
	"github.com/influxdata/influxdb/tcp"
	"github.com/influxdata/influxdb/tsdb"
	"github.com/influxdata/platform/storage/reads"
	client "github.com/influxdata/usage-client/v1"
	"go.uber.org/zap"
	_ "github.com/influxdata/influxdb/tsdb/engine"
	_ "github.com/influxdata/influxdb/tsdb/index"
	"github.com/angopher/chronus/coordinator"
	"github.com/angopher/chronus/services/controller"
	"github.com/angopher/chronus/services/hh"
	imeta "github.com/angopher/chronus/services/meta"
)
var startTime time.Time
func init() {
	startTime = time.Now().UTC()
}
type BuildInfo struct {
	Version string
	Commit  string
	Branch  string
	Time    string
}
type Server struct {
	buildInfo BuildInfo
	err     chan error
	closing chan struct{}
	BindAddress string
	Listener    net.Listener
	Logger *zap.Logger
	Node              *influxdb.Node
	ClusterMetaClient *coordinator.ClusterMetaClient
	TSDBStore     *tsdb.Store
	QueryExecutor *query.Executor
	PointsWriter  *coordinator.PointsWriter
	ShardWriter   *coordinator.ShardWriter
	HintedHandoff *hh.Service
	Subscriber    *subscriber.Service
	Services []Service
	SnapshotterService *snapshotter.Service
	ClusterService     *coordinator.Service
	ControllerService  *controller.Service
	Monitor *monitor.Monitor
	reportingDisabled bool
	CPUProfile string
	MemProfile string
	httpAPIAddr string
	httpUseTLS bool
	tcpAddr string
	config *Config
}
func updateTLSConfig(into **tls.Config, with *tls.Config) {
	if with != nil && into != nil && *into == nil {
		*into = with
	}
}
func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {
	tlsConfig, err := c.TLS.Parse()
	if err != nil {
		return nil, fmt.Errorf("tls configuration: %v", err)
	}
	updateTLSConfig(&c.HTTPD.TLS, tlsConfig)
	updateTLSConfig(&c.Subscriber.TLS, tlsConfig)
	for i := range c.OpenTSDBInputs {
		updateTLSConfig(&c.OpenTSDBInputs[i].TLS, tlsConfig)
	}
	if err := os.MkdirAll(c.Meta.Dir, 0777); err != nil {
		return nil, fmt.Errorf("mkdir all: %s", err)
	}
	oldPath := filepath.Join(filepath.Dir(c.Meta.Dir), "node.json")
	newPath := filepath.Join(c.Meta.Dir, "node.json")
	if _, err := os.Stat(oldPath); err == nil {
		if err := os.Rename(oldPath, newPath); err != nil {
			return nil, err
		}
	}
	node, err := influxdb.LoadNode(c.Meta.Dir)
	if err != nil {
		if !os.IsNotExist(err) {
			return nil, err
		}
		node = influxdb.NewNode(c.Meta.Dir)
	}
	var nodeID uint64 = 0
	if node != nil {
		nodeID = node.ID
	}
	if err := raftDBExists(c.Meta.Dir); err != nil {
		return nil, err
	}
	bind := c.BindAddress
	s := &Server{
		buildInfo: *buildInfo,
		err:       make(chan error),
		closing:   make(chan struct{}),
		BindAddress: bind,
		Node: node,
		Logger: logger.New(os.Stderr),
		ClusterMetaClient: coordinator.NewMetaClient(c.Meta, c.Coordinator, nodeID),
		reportingDisabled: c.ReportingDisabled,
		httpAPIAddr: c.HTTPD.BindAddress,
		httpUseTLS:  c.HTTPD.HTTPSEnabled,
		tcpAddr:     bind,
		config: c,
	}
	s.ClusterMetaClient.WithLogger(s.Logger)
	s.Monitor = monitor.New(s, c.Monitor)
	s.config.registerDiagnostics(s.Monitor)
	if err := s.ClusterMetaClient.Open(); err != nil {
		return nil, err
	}
	wait := s.ClusterMetaClient.WaitForDataChanged()
	go s.ClusterMetaClient.RunSyncLoop()
	select {
	case <-time.After(5 * time.Second):
		panic("sync meta data failed")
	case <-wait:
	}
	n, err := s.ClusterMetaClient.DataNode(nodeID)
	if err != nil {
		n, err = s.ClusterMetaClient.CreateDataNode(s.httpAPIAddr, s.tcpAddr)
		if err != nil {
			log.Printf("Unable to create data node. err: %s", err.Error())
			return nil, err
		}
	}
	s.Node.ID = n.ID
	s.ClusterMetaClient.NodeID = n.ID
	if err := s.Node.Save(); err != nil {
		return nil, err
	}
	s.TSDBStore = tsdb.NewStore(c.Data.Dir)
	s.TSDBStore.EngineOptions.Config = c.Data
	s.TSDBStore.EngineOptions.EngineVersion = c.Data.Engine
	s.TSDBStore.EngineOptions.IndexVersion = c.Data.Index
	s.Subscriber = subscriber.NewService(c.Subscriber)
	s.ShardWriter = coordinator.NewShardWriter(time.Duration(s.config.Coordinator.WriteTimeout), s.config.Coordinator.PoolMaxConnections)
	s.HintedHandoff = hh.NewService(c.HintedHandoff, s.ShardWriter, s.ClusterMetaClient)
	s.HintedHandoff.Monitor = s.Monitor
	s.PointsWriter = coordinator.NewPointsWriter()
	s.PointsWriter.WriteTimeout = time.Duration(c.Coordinator.WriteTimeout)
	s.PointsWriter.TSDBStore = s.TSDBStore
	s.PointsWriter.ShardWriter = s.ShardWriter
	s.PointsWriter.Node = s.Node
	s.PointsWriter.HintedHandoff = s.HintedHandoff
	clusterExecutor := coordinator.NewClusterExecutor(s.Node, s.TSDBStore, s.ClusterMetaClient, s.config.Coordinator)
	clusterExecutor.WithLogger(s.Logger)
	s.QueryExecutor = query.NewExecutor()
	s.QueryExecutor.StatementExecutor = &influxdb_coordinator.StatementExecutor{
		MetaClient:  s.ClusterMetaClient,
		TaskManager: clusterExecutor,
		TSDBStore:   clusterExecutor,
		ShardMapper: &coordinator.ClusterShardMapper{
			MetaClient:      s.ClusterMetaClient,
			Node:            s.Node,
			ClusterExecutor: clusterExecutor,
		},
		Monitor:           s.Monitor,
		PointsWriter:      s.PointsWriter,
		MaxSelectPointN:   c.Coordinator.MaxSelectPointN,
		MaxSelectSeriesN:  c.Coordinator.MaxSelectSeriesN,
		MaxSelectBucketsN: c.Coordinator.MaxSelectBucketsN,
	}
	s.QueryExecutor.TaskManager.QueryTimeout = time.Duration(c.Coordinator.QueryTimeout)
	s.QueryExecutor.TaskManager.LogQueriesAfter = time.Duration(c.Coordinator.LogQueriesAfter)
	s.QueryExecutor.TaskManager.MaxConcurrentQueries = c.Coordinator.MaxConcurrentQueries
	clusterExecutor.TaskManager = s.QueryExecutor.TaskManager
	s.Monitor.Version = s.buildInfo.Version
	s.Monitor.Commit = s.buildInfo.Commit
	s.Monitor.Branch = s.buildInfo.Branch
	s.Monitor.BuildTime = s.buildInfo.Time
	s.Monitor.PointsWriter = (*monitorPointsWriter)(s.PointsWriter)
	return s, nil
}
func (s *Server) Statistics(tags map[string]string) []models.Statistic {
	var statistics []models.Statistic
	statistics = append(statistics, s.QueryExecutor.Statistics(tags)...)
	statistics = append(statistics, s.TSDBStore.Statistics(tags)...)
	statistics = append(statistics, s.PointsWriter.Statistics(tags)...)
	statistics = append(statistics, s.Subscriber.Statistics(tags)...)
	for _, srv := range s.Services {
		if m, ok := srv.(monitor.Reporter); ok {
			statistics = append(statistics, m.Statistics(tags)...)
		}
	}
	return statistics
}
func (s *Server) appendClusterService(c coordinator.Config) {
	srv := coordinator.NewService(c)
	srv.TaskManager = s.QueryExecutor.TaskManager
	srv.TSDBStore = s.TSDBStore
	srv.Node = s.Node
	srv.MetaClient = s.ClusterMetaClient
	srv.WithLogger(s.Logger)
	s.Services = append(s.Services, srv)
	s.ClusterService = srv
}
func (s *Server) appendSnapshotterService() {
	srv := snapshotter.NewService()
	srv.TSDBStore = s.TSDBStore
	srv.MetaClient = s.ClusterMetaClient
	s.Services = append(s.Services, srv)
	s.SnapshotterService = srv
}
func (s *Server) SetLogOutput(w io.Writer) {
	s.Logger = logger.New(w)
}
func (s *Server) appendMonitorService() {
	s.Services = append(s.Services, s.Monitor)
}
func (s *Server) appendRetentionPolicyService(c retention.Config) {
	if !c.Enabled {
		return
	}
	srv := retention.NewService(c)
	srv.MetaClient = s.ClusterMetaClient
	srv.TSDBStore = s.TSDBStore
	s.Services = append(s.Services, srv)
}
func (s *Server) appendHTTPDService(c httpd.Config) {
	if !c.Enabled {
		return
	}
	srv := httpd.NewService(c)
	srv.Handler.MetaClient = s.ClusterMetaClient
	authorizer := &imeta.Authorizer{}
	srv.Handler.QueryAuthorizer = authorizer
	srv.Handler.WriteAuthorizer = authorizer
	srv.Handler.QueryExecutor = s.QueryExecutor
	srv.Handler.Monitor = s.Monitor
	srv.Handler.PointsWriter = s.PointsWriter
	srv.Handler.Version = s.buildInfo.Version
	srv.Handler.BuildType = "OSS"
	ss := storage.NewStore(s.TSDBStore, s.ClusterMetaClient)
	srv.Handler.Store = ss
	srv.Handler.Controller = control.NewController(s.ClusterMetaClient, reads.NewReader(ss), authorizer, c.AuthEnabled, s.Logger)
	s.Services = append(s.Services, srv)
}
func (s *Server) appendCollectdService(c collectd.Config) {
	if !c.Enabled {
		return
	}
	srv := collectd.NewService(c)
	srv.MetaClient = s.ClusterMetaClient
	srv.PointsWriter = s.PointsWriter
	s.Services = append(s.Services, srv)
}
func (s *Server) appendOpenTSDBService(c opentsdb.Config) error {
	if !c.Enabled {
		return nil
	}
	srv, err := opentsdb.NewService(c)
	if err != nil {
		return err
	}
	srv.PointsWriter = s.PointsWriter
	srv.MetaClient = s.ClusterMetaClient
	s.Services = append(s.Services, srv)
	return nil
}
func (s *Server) appendGraphiteService(c graphite.Config) error {
	if !c.Enabled {
		return nil
	}
	srv, err := graphite.NewService(c)
	if err != nil {
		return err
	}
	srv.PointsWriter = s.PointsWriter
	srv.MetaClient = s.ClusterMetaClient
	srv.Monitor = s.Monitor
	s.Services = append(s.Services, srv)
	return nil
}
func (s *Server) appendPrecreatorService(c precreator.Config) error {
	if !c.Enabled {
		return nil
	}
	srv := precreator.NewService(c)
	srv.MetaClient = s.ClusterMetaClient
	s.Services = append(s.Services, srv)
	return nil
}
func (s *Server) appendUDPService(c udp.Config) {
	if !c.Enabled {
		return
	}
	srv := udp.NewService(c)
	srv.PointsWriter = s.PointsWriter
	srv.MetaClient = s.ClusterMetaClient
	s.Services = append(s.Services, srv)
}
func (s *Server) appendContinuousQueryService(c continuous_querier.Config) {
	if !c.Enabled {
		return
	}
	srv := continuous_querier.NewService(c)
	srv.MetaClient = s.ClusterMetaClient
	srv.QueryExecutor = s.QueryExecutor
	srv.Monitor = s.Monitor
	s.Services = append(s.Services, srv)
}
func (s *Server) appendControllerService(c controller.Config) {
	if !c.Enabled {
		return
	}
	srv := controller.NewService(c)
	srv.MetaClient = s.ClusterMetaClient
	srv.Node = s.Node
	srv.TSDBStore = s.TSDBStore
	shardCopier := &controller.ShardCopier{
		Manager:    controller.NewCopyManager(c.MaxShardCopyTasks),
		MetaClient: s.ClusterMetaClient,
		TSDBStore:  s.TSDBStore,
		Node:       s.Node,
	}
	shardCopier.WithLogger(s.Logger)
	srv.ShardCopier = shardCopier
	s.ControllerService = srv
	s.Services = append(s.Services, srv)
}
func (s *Server) Err() <-chan error { return s.err }
func (s *Server) Open() error {
	startProfile(s.CPUProfile, s.MemProfile)
	ln, err := net.Listen("tcp", s.BindAddress)
	if err != nil {
		return fmt.Errorf("listen: %s", err)
	}
	s.Listener = ln
	mux := tcp.NewMux()
	go mux.Serve(ln)
	s.appendMonitorService()
	s.appendPrecreatorService(s.config.Precreator)
	s.appendClusterService(s.config.Coordinator)
	s.appendSnapshotterService()
	s.appendContinuousQueryService(s.config.ContinuousQuery)
	s.appendHTTPDService(s.config.HTTPD)
	s.appendRetentionPolicyService(s.config.Retention)
	s.appendControllerService(s.config.Controller)
	for _, i := range s.config.GraphiteInputs {
		if err := s.appendGraphiteService(i); err != nil {
			return err
		}
	}
	for _, i := range s.config.CollectdInputs {
		s.appendCollectdService(i)
	}
	for _, i := range s.config.OpenTSDBInputs {
		if err := s.appendOpenTSDBService(i); err != nil {
			return err
		}
	}
	for _, i := range s.config.UDPInputs {
		s.appendUDPService(i)
	}
	s.Subscriber.MetaClient = s.ClusterMetaClient
	s.PointsWriter.MetaClient = s.ClusterMetaClient
	s.ShardWriter.MetaClient = s.ClusterMetaClient
	s.Monitor.MetaClient = s.ClusterMetaClient
	s.ClusterService.Listener = mux.Listen(coordinator.MuxHeader)
	s.SnapshotterService.Listener = mux.Listen(snapshotter.MuxHeader)
	s.ControllerService.Listener = mux.Listen(controller.MuxHeader)
	s.TSDBStore.WithLogger(s.Logger)
	if s.config.Data.QueryLogEnabled {
		s.QueryExecutor.WithLogger(s.Logger)
	}
	s.PointsWriter.WithLogger(s.Logger)
	s.Subscriber.WithLogger(s.Logger)
	for _, svc := range s.Services {
		svc.WithLogger(s.Logger)
	}
	s.SnapshotterService.WithLogger(s.Logger)
	s.Monitor.WithLogger(s.Logger)
	if err := s.TSDBStore.Open(); err != nil {
		return fmt.Errorf("open tsdb store: %s", err)
	}
	if err := s.Subscriber.Open(); err != nil {
		return fmt.Errorf("open subscriber: %s", err)
	}
	if err := s.HintedHandoff.Open(); err != nil {
		return fmt.Errorf("open hinted handoff: %s", err)
	}
	if err := s.PointsWriter.Open(); err != nil {
		return fmt.Errorf("open points writer: %s", err)
	}
	for _, service := range s.Services {
		if err := service.Open(); err != nil {
			return fmt.Errorf("open service: %s", err)
		}
	}
	if !s.reportingDisabled {
		go s.startServerReporting()
	}
	return nil
}
func (s *Server) Close() error {
	stopProfile()
	if s.Listener != nil {
		s.Listener.Close()
	}
	for _, service := range s.Services {
		service.Close()
	}
	s.config.deregisterDiagnostics(s.Monitor)
	if s.PointsWriter != nil {
		s.PointsWriter.Close()
	}
	if s.QueryExecutor != nil {
		s.QueryExecutor.Close()
	}
	if s.TSDBStore != nil {
		s.TSDBStore.Close()
	}
	if s.Subscriber != nil {
		s.Subscriber.Close()
	}
	close(s.closing)
	return nil
}
func (s *Server) startServerReporting() {
	s.reportServer()
	ticker := time.NewTicker(24 * time.Hour)
	defer ticker.Stop()
	for {
		select {
		case <-s.closing:
			return
		case <-ticker.C:
			s.reportServer()
		}
	}
}