func (w *PointsWriter) WritePointsInto(p *influxdb_coordinator.IntoWriteRequest) error {
	return w.WritePointsPrivileged(p.Database, p.RetentionPolicy, models.ConsistencyLevelOne, p.Points)
}
WritePointsInto is a copy of WritePoints that uses a tsdb structure instead of
a cluster structure for information. This is to avoid a circular dependency.
package coordinator
import (
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/influxdata/influxdb"
	influxdb_coordinator "github.com/influxdata/influxdb/coordinator"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/tsdb"
	"go.uber.org/zap"
	"github.com/influxdata/influxdb/services/meta"
)
const (
	statWriteReq            = "req"
	statPointWriteReq       = "pointReq"
	statPointWriteReqLocal  = "pointReqLocal"
	statPointWriteReqRemote = "pointReqRemote"
	statWriteOK             = "writeOk"
	statWriteDrop           = "writeDrop"
	statWritePartial        = "writePartial"
	statWriteTimeout        = "writeTimeout"
	statWriteErr            = "writeError"
	statWritePointReqHH     = "pointReqHH"
	statSubWriteOK          = "subWriteOk"
	statSubWriteDrop        = "subWriteDrop"
)
var (
	ErrTimeout = errors.New("timeout")
	ErrPartialWrite = errors.New("partial write")
	ErrWriteFailed = errors.New("write failed")
)
type PointsWriter struct {
	mu           sync.RWMutex
	closing      chan struct{}
	WriteTimeout time.Duration
	Logger       *zap.Logger
	Node *influxdb.Node
	HintedHandoff interface {
		WriteShard(shardID, ownerID uint64, points []models.Point) error
	}
	MetaClient interface {
		Database(name string) (di *meta.DatabaseInfo)
		RetentionPolicy(database, policy string) (*meta.RetentionPolicyInfo, error)
		CreateShardGroup(database, policy string, timestamp time.Time) (*meta.ShardGroupInfo, error)
	}
	TSDBStore interface {
		CreateShard(database, retentionPolicy string, shardID uint64, enabled bool) error
		WriteToShard(shardID uint64, points []models.Point) error
	}
	ShardWriter interface {
		WriteShard(shardID, ownerID uint64, points []models.Point) error
	}
	subPoints []chan<- *WritePointsRequest
	stats *WriteStatistics
}
func NewPointsWriter() *PointsWriter {
	return &PointsWriter{
		closing:      make(chan struct{}),
		WriteTimeout: DefaultWriteTimeout,
		Logger:       zap.NewNop(),
		stats:        &WriteStatistics{},
	}
}
type ShardMapping struct {
	n       int
	Points  map[uint64][]models.Point   
	Shards  map[uint64]*meta.ShardInfo  
	Dropped []models.Point              
}
func NewShardMapping(n int) *ShardMapping {
	return &ShardMapping{
		n:      n,
		Points: map[uint64][]models.Point{},
		Shards: map[uint64]*meta.ShardInfo{},
	}
}
func (s *ShardMapping) MapPoint(shardInfo *meta.ShardInfo, p models.Point) {
	if cap(s.Points[shardInfo.ID]) < s.n {
		s.Points[shardInfo.ID] = make([]models.Point, 0, s.n)
	}
	s.Points[shardInfo.ID] = append(s.Points[shardInfo.ID], p)
	s.Shards[shardInfo.ID] = shardInfo
}
func (w *PointsWriter) Open() error {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.closing = make(chan struct{})
	return nil
}
func (w *PointsWriter) Close() error {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.closing != nil {
		close(w.closing)
	}
	if w.subPoints != nil {
		w.subPoints = nil
	}
	return nil
}
func (w *PointsWriter) AddWriteSubscriber(c chan<- *WritePointsRequest) {
	w.subPoints = append(w.subPoints, c)
}
func (w *PointsWriter) WithLogger(log *zap.Logger) {
	w.Logger = log.With(zap.String("service", "write"))
}
type WriteStatistics struct {
	WriteReq            int64
	PointWriteReq       int64
	PointWriteReqLocal  int64
	PointWriteReqRemote int64
	WriteOK             int64
	WriteDropped        int64
	WriteTimeout        int64
	WritePartial        int64
	WritePointReqHH     int64
	WriteErr            int64
	SubWriteOK          int64
	SubWriteDrop        int64
}
func (w *PointsWriter) Statistics(tags map[string]string) []models.Statistic {
	return []models.Statistic{{
		Name: "write",
		Tags: tags,
		Values: map[string]interface{}{
			statWriteReq:            atomic.LoadInt64(&w.stats.WriteReq),
			statPointWriteReq:       atomic.LoadInt64(&w.stats.PointWriteReq),
			statPointWriteReqLocal:  atomic.LoadInt64(&w.stats.PointWriteReqLocal),
			statPointWriteReqRemote: atomic.LoadInt64(&w.stats.PointWriteReqRemote),
			statWriteOK:             atomic.LoadInt64(&w.stats.WriteOK),
			statWriteDrop:           atomic.LoadInt64(&w.stats.WriteDropped),
			statWriteTimeout:        atomic.LoadInt64(&w.stats.WriteTimeout),
			statWritePartial:        atomic.LoadInt64(&w.stats.WritePartial),
			statWritePointReqHH:     atomic.LoadInt64(&w.stats.WritePointReqHH),
			statWriteErr:            atomic.LoadInt64(&w.stats.WriteErr),
			statSubWriteOK:          atomic.LoadInt64(&w.stats.SubWriteOK),
			statSubWriteDrop:        atomic.LoadInt64(&w.stats.SubWriteDrop),
		},
	}}
}
func (w *PointsWriter) MapShards(wp *WritePointsRequest) (*ShardMapping, error) {
	rp, err := w.MetaClient.RetentionPolicy(wp.Database, wp.RetentionPolicy)
	if err != nil {
		return nil, err
	} else if rp == nil {
		return nil, influxdb.ErrRetentionPolicyNotFound(wp.RetentionPolicy)
	}
	list := make(sgList, 0, 8)
	min := time.Unix(0, models.MinNanoTime)
	if rp.Duration > 0 {
		min = time.Now().Add(-rp.Duration)
	}
	for _, p := range wp.Points {
		if p.Time().Before(min) || list.Covers(p.Time()) {
			continue
		}
		sg, err := w.MetaClient.CreateShardGroup(wp.Database, wp.RetentionPolicy, p.Time())
		if err != nil {
			return nil, err
		}
		if sg == nil {
			return nil, errors.New("nil shard group")
		}
		list = list.Append(*sg)
	}
	mapping := NewShardMapping(len(wp.Points))
	for _, p := range wp.Points {
		sg := list.ShardGroupAt(p.Time())
		if sg == nil {
			mapping.Dropped = append(mapping.Dropped, p)
			atomic.AddInt64(&w.stats.WriteDropped, 1)
			continue
		}
		sh := sg.ShardFor(p.HashID())
		mapping.MapPoint(&sh, p)
	}
	return mapping, nil
}
type sgList meta.ShardGroupInfos
func (l sgList) Covers(t time.Time) bool {
	if len(l) == 0 {
		return false
	}
	return l.ShardGroupAt(t) != nil
}
func (l sgList) ShardGroupAt(t time.Time) *meta.ShardGroupInfo {
	idx := sort.Search(len(l), func(i int) bool { return l[i].EndTime.After(t) })
	if idx == len(l) || t.Before(l[idx].StartTime) {
		return nil
	}
	return &l[idx]
}
func (l sgList) Append(sgi meta.ShardGroupInfo) sgList {
	next := append(l, sgi)
	sort.Sort(meta.ShardGroupInfos(next))
	return next
}