func (s *Service) handleConn(conn net.Conn) {
	closing := make(chan struct{})
	defer close(closing)
	go func() {
		select {
		case <-closing:
		case <-s.closing:
		}
		conn.Close()
	}()
	s.Logger.Info(fmt.Sprintf("accept remote connection from %+v", conn.RemoteAddr()))
	defer func() {
		s.Logger.Info(fmt.Sprintf("close remote connection from %+v", conn.RemoteAddr()))
	}()
	for {
		typ, err := ReadType(conn)
		if err != nil {
			if strings.HasSuffix(err.Error(), "EOF") {
				return
			}
			s.Logger.Error("unable to read type", zap.Error(err))
			return
		}
		switch typ {
		case writeShardRequestMessage:
			buf, err := ReadLV(conn)
			if err != nil {
				s.Logger.Error("unable to read length-value", zap.Error(err))
				return
			}
			atomic.AddInt64(&s.stats.WriteShardReq, 1)
			err = s.processWriteShardRequest(buf)
			if err != nil {
				s.Logger.Error("process write shard error", zap.Error(err))
			}
			s.writeShardResponse(conn, err)
		case executeStatementRequestMessage:
			buf, err := ReadLV(conn)
			if err != nil {
				s.Logger.Error("unable to read length-value", zap.Error(err))
				return
			}
			err = s.processExecuteStatementRequest(buf)
			if err != nil {
				s.Logger.Error("process execute statement error", zap.Error(err))
			}
			s.writeShardResponse(conn, err)
		case createIteratorRequestMessage:
			atomic.AddInt64(&s.stats.CreateIteratorReq, 1)
			s.processCreateIteratorRequest(conn)
			return
		case fieldDimensionsRequestMessage:
			atomic.AddInt64(&s.stats.FieldDimensionsReq, 1)
			s.processFieldDimensionsRequest(conn)
			return
		case tagKeysRequestMessage:
			atomic.AddInt64(&s.stats.TagKeysReq, 1)
			s.processTagKeysRequest(conn)
		case tagValuesRequestMessage:
			atomic.AddInt64(&s.stats.TagValuesReq, 1)
			s.processTagValuesRequest(conn)
			return
		case measurementNamesRequestMessage:
			atomic.AddInt64(&s.stats.MeasurementNamesReq, 1)
			s.processMeasurementNamesRequest(conn)
			return
		case seriesCardinalityRequestMessage:
			atomic.AddInt64(&s.stats.SeriesCardinalityReq, 1)
			s.processSeriesCardinalityRequest(conn)
			return
		case deleteSeriesRequestMessage:
			s.processDeleteSeriesRequest(conn)
			return
		case deleteDatabaseRequestMessage:
			s.processDeleteDatabaseRequest(conn)
			return
		case deleteMeasurementRequestMessage:
			s.processDeleteMeasurementRequest(conn)
			return
		case iteratorCostRequestMessage:
			atomic.AddInt64(&s.stats.IteratorCostReq, 1)
			s.processIteratorCostRequest(conn)
			return
		case mapTypeRequestMessage:
			atomic.AddInt64(&s.stats.MapTypeReq, 1)
			s.processMapTypeRequest(conn)
			return
		case executeTaskManagerRequestMessage:
			s.processTaskManagerRequest(conn)
			return
		default:
			s.Logger.Error("cluster service message type not found", zap.Uint8("type", typ))
		}
	}
}
handleConn services an individual TCP connection.
package coordinator
import (
	"context"
	"encoding"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"net"
	"runtime/debug"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/influxdata/influxdb"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/pkg/tracing"
	"github.com/influxdata/influxdb/query"
	"github.com/influxdata/influxdb/tsdb"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
	"github.com/influxdata/influxdb/services/meta"
)
const MaxMessageSize = 1024 * 1024 * 1024  
const MuxHeader = 2
const (
	writeShardReq       = "writeShardReq"
	writeShardPointsReq = "writeShardPointsReq"
	writeShardFail      = "writeShardFail"
	createIteratorReq  = "createIteratorReq"
	createIteratorFail = "createIteratorFail"
	fieldDimensionsReq  = "fieldDimensionsReq"
	fieldDimensionsFail = "fieldDimensionsFail"
	tagKeysReq  = "tagKeysReq"
	tagKeysFail = "tagKeysFail"
	tagValuesReq  = "tagValuesReq"
	tagValuesFail = "tagValuesFail"
	measurementNamesReq  = "measurementNamesReq"
	measurementNamesFail = "measurementNamesFail"
	seriesCardinalityReq  = "seriesCardinalityReq"
	seriesCardinalityFail = "seriesCardinalityFail"
	iteratorCostReq  = "iteratorCostReq"
	iteratorCostFail = "iteratorCostFail"
	mapTypeReq  = "mapTypeReq"
	mapTypeFail = "mapTypeFail"
)
type Service struct {
	mu      sync.RWMutex
	Node    *influxdb.Node
	wg      sync.WaitGroup
	closing chan struct{}
	Listener net.Listener
	MetaClient interface {
		ShardOwner(shardID uint64) (string, string, *meta.ShardGroupInfo)
	}
	TSDBStore   TSDBStore
	TaskManager *query.TaskManager
	Logger *zap.Logger
	stats  *InternalServiceStatistics
}
func NewService(c Config) *Service {
	return &Service{
		closing: make(chan struct{}),
		stats:  &InternalServiceStatistics{},
		Logger: zap.NewNop(),
	}
}
func (s *Service) Open() error {
	s.Logger.Info("Starting cluster service")
	s.wg.Add(1)
	go s.serve()
	return nil
}
func (s *Service) WithLogger(log *zap.Logger) {
	s.Logger = log.With(zap.String("service", "cluster"))
}
type InternalServiceStatistics struct {
	WriteShardReq       int64
	WriteShardPointsReq int64
	WriteShardFail      int64
	CreateIteratorReq  int64
	CreateIteratorFail int64
	FieldDimensionsReq  int64
	FieldDimensionsFail int64
	TagKeysReq  int64
	TagKeysFail int64
	TagValuesReq  int64
	TagValuesFail int64
	MeasurementNamesReq  int64
	MeasurementNamesFail int64
	SeriesCardinalityReq  int64
	SeriesCardinalityFail int64
	IteratorCostReq  int64
	IteratorCostFail int64
	MapTypeReq  int64
	MapTypeFail int64
}
func (w *Service) Statistics(tags map[string]string) []models.Statistic {
	return []models.Statistic{{
		Name: "coordinator_service",
		Tags: tags,
		Values: map[string]interface{}{
			writeShardReq:       atomic.LoadInt64(&w.stats.WriteShardReq),
			writeShardPointsReq: atomic.LoadInt64(&w.stats.WriteShardPointsReq),
			writeShardFail:      atomic.LoadInt64(&w.stats.WriteShardFail),
			createIteratorReq:  atomic.LoadInt64(&w.stats.CreateIteratorReq),
			createIteratorFail: atomic.LoadInt64(&w.stats.CreateIteratorFail),
			fieldDimensionsReq:  atomic.LoadInt64(&w.stats.FieldDimensionsReq),
			fieldDimensionsFail: atomic.LoadInt64(&w.stats.FieldDimensionsFail),
			tagKeysReq:  atomic.LoadInt64(&w.stats.TagKeysReq),
			tagKeysFail: atomic.LoadInt64(&w.stats.TagKeysFail),
			tagValuesReq:  atomic.LoadInt64(&w.stats.TagValuesReq),
			tagValuesFail: atomic.LoadInt64(&w.stats.TagValuesFail),
			measurementNamesReq:  atomic.LoadInt64(&w.stats.MeasurementNamesReq),
			measurementNamesFail: atomic.LoadInt64(&w.stats.MeasurementNamesFail),
			seriesCardinalityReq:  atomic.LoadInt64(&w.stats.SeriesCardinalityReq),
			seriesCardinalityFail: atomic.LoadInt64(&w.stats.SeriesCardinalityFail),
			iteratorCostReq:  atomic.LoadInt64(&w.stats.IteratorCostReq),
			iteratorCostFail: atomic.LoadInt64(&w.stats.IteratorCostFail),
			mapTypeReq:  atomic.LoadInt64(&w.stats.MapTypeReq),
			mapTypeFail: atomic.LoadInt64(&w.stats.MapTypeFail),
		},
	}}
}
func (s *Service) serve() {
	defer s.wg.Done()
	for {
		select {
		case <-s.closing:
			return
		default:
		}
		conn, err := s.Listener.Accept()
		if err != nil {
			if strings.Contains(err.Error(), "connection closed") {
				s.Logger.Error("cluster service accept fail", zap.Error(err))
				return
			}
			s.Logger.Error("accept error", zap.Error(err))
			continue
		}
		s.wg.Add(1)
		go func() {
			defer func() {
				s.wg.Done()
				if err := recover(); err != nil {
					buf := debug.Stack()
					s.Logger.Error("recover from panic", zap.String("stack", string(buf)))
				}
			}()
			s.handleConn(conn)
		}()
	}
}
func (s *Service) Close() error {
	if s.Listener != nil {
		s.Listener.Close()
	}
	close(s.closing)
	s.wg.Wait()
	return nil
}