func (et *ExecutingTask) registerOutput(name string, o Output) {
	et.outputs[name] = o
}
Register a named output.
package kapacitor
import (
	"bytes"
	"errors"
	"fmt"
	"log"
	"math/rand"
	"sync"
	"time"
	"github.com/influxdata/kapacitor/edge"
	"github.com/influxdata/kapacitor/pipeline"
)
type TaskType int
const (
	StreamTask TaskType = iota
	BatchTask
)
func (t TaskType) String() string {
	switch t {
	case StreamTask:
		return "stream"
	case BatchTask:
		return "batch"
	default:
		return "unknown"
	}
}
func (t TaskType) MarshalText() ([]byte, error) {
	return []byte(t.String()), nil
}
func (t *TaskType) UnmarshalText(text []byte) error {
	switch string(text) {
	case "stream":
		*t = StreamTask
	case "batch":
		*t = BatchTask
	default:
		return fmt.Errorf("unknown task type %s", string(text))
	}
	return nil
}
type DBRP struct {
	Database        string `json:"db"`
	RetentionPolicy string `json:"rp"`
}
func CreateDBRPMap(dbrps []DBRP) map[DBRP]bool {
	dbMap := make(map[DBRP]bool, len(dbrps))
	for _, dbrp := range dbrps {
		dbMap[dbrp] = true
	}
	return dbMap
}
func (d DBRP) String() string {
	return fmt.Sprintf("%q.%q", d.Database, d.RetentionPolicy)
}
type Task struct {
	ID               string
	Pipeline         *pipeline.Pipeline
	Type             TaskType
	DBRPs            []DBRP
	SnapshotInterval time.Duration
}
func (t *Task) Dot() []byte {
	return t.Pipeline.Dot(t.ID)
}
func (t *Task) Measurements() []string {
	measurements := make([]string, 0)
	_ = t.Pipeline.Walk(func(node pipeline.Node) error {
		switch streamNode := node.(type) {
		case *pipeline.FromNode:
			measurements = append(measurements, streamNode.Measurement)
		}
		return nil
	})
	return measurements
}
type ExecutingTask struct {
	tm      *TaskMaster
	Task    *Task
	source  Node
	outputs map[string]Output
	lookup   map[pipeline.ID]Node
	nodes    []Node
	stopping chan struct{}
	wg       sync.WaitGroup
	logger   *log.Logger
	tmu        sync.RWMutex
	throughput float64
}
func NewExecutingTask(tm *TaskMaster, t *Task) (*ExecutingTask, error) {
	l := tm.LogService.NewLogger(fmt.Sprintf("[task:%s] ", t.ID), log.LstdFlags)
	et := &ExecutingTask{
		tm:      tm,
		Task:    t,
		outputs: make(map[string]Output),
		lookup:  make(map[pipeline.ID]Node),
		logger:  l,
	}
	err := et.link()
	if err != nil {
		return nil, err
	}
	return et, nil
}
func (et *ExecutingTask) walk(f func(n Node) error) error {
	for _, n := range et.nodes {
		err := f(n)
		if err != nil {
			return err
		}
	}
	return nil
}
func (et *ExecutingTask) rwalk(f func(n Node) error) error {
	for i := len(et.nodes) - 1; i >= 0; i-- {
		err := f(et.nodes[i])
		if err != nil {
			return err
		}
	}
	return nil
}
func (et *ExecutingTask) link() error {
	err := et.Task.Pipeline.Walk(func(n pipeline.Node) error {
		l := et.tm.LogService.NewLogger(
			fmt.Sprintf("[%s:%s] ", et.Task.ID, n.Name()),
			log.LstdFlags,
		)
		en, err := et.createNode(n, l)
		if err != nil {
			return err
		}
		et.lookup[n.ID()] = en
		et.nodes = append(et.nodes, en)
		for _, p := range n.Parents() {
			ep := et.lookup[p.ID()]
			err := ep.linkChild(en)
			if err != nil {
				return err
			}
		}
		return err
	})
	if err != nil {
		return err
	}
	et.source = et.nodes[0]
	return nil
}
func (et *ExecutingTask) start(ins []edge.StatsEdge, snapshot *TaskSnapshot) error {
	for _, in := range ins {
		et.source.addParentEdge(in)
	}
	validSnapshot := false
	if snapshot != nil {
		err := et.walk(func(n Node) error {
			_, ok := snapshot.NodeSnapshots[n.Name()]
			if !ok {
				return fmt.Errorf("task pipeline changed not using snapshot")
			}
			return nil
		})
		validSnapshot = err == nil
	}
	err := et.walk(func(n Node) error {
		if validSnapshot {
			n.start(snapshot.NodeSnapshots[n.Name()])
		} else {
			n.start(nil)
		}
		return nil
	})
	if err != nil {
		return err
	}
	et.stopping = make(chan struct{})
	if et.Task.SnapshotInterval > 0 {
		et.wg.Add(1)
		go et.runSnapshotter()
	}
	et.wg.Add(1)
	go et.calcThroughput()
	return nil
}
func (et *ExecutingTask) stop() (err error) {
	close(et.stopping)
	_ = et.walk(func(n Node) error {
		n.stop()
		e := n.Wait()
		if e != nil {
			err = e
		}
		return nil
	})
	et.wg.Wait()
	return
}
var ErrWrongTaskType = errors.New("wrong task type")
func (et *ExecutingTask) StartBatching() error {
	if et.Task.Type != BatchTask {
		return ErrWrongTaskType
	}
	batcher := et.source.(*BatchNode)
	err := et.checkDBRPs(batcher)
	if err != nil {
		batcher.Abort()
		return err
	}
	batcher.Start()
	return nil
}
func (et *ExecutingTask) BatchCount() (int, error) {
	if et.Task.Type != BatchTask {
		return 0, ErrWrongTaskType
	}
	batcher := et.source.(*BatchNode)
	return batcher.Count(), nil
}
func (et *ExecutingTask) BatchQueries(start, stop time.Time) ([]BatchQueries, error) {
	if et.Task.Type != BatchTask {
		return nil, ErrWrongTaskType
	}
	batcher := et.source.(*BatchNode)
	err := et.checkDBRPs(batcher)
	if err != nil {
		return nil, err
	}
	return batcher.Queries(start, stop)
}
func (et *ExecutingTask) checkDBRPs(batcher *BatchNode) error {
	dbMap := CreateDBRPMap(et.Task.DBRPs)
	dbrps, err := batcher.DBRPs()
	if err != nil {
		return err
	}
	for _, dbrp := range dbrps {
		if !dbMap[dbrp] {
			return fmt.Errorf("batch query is not allowed to request data from %v", dbrp)
		}
	}
	return nil
}
func (et *ExecutingTask) StopStats() {
	_ = et.walk(func(n Node) error {
		if s, ok := n.(*StatsNode); ok {
			s.stopStats()
		}
		return nil
	})
}
func (et *ExecutingTask) Wait() error {
	return et.rwalk(func(n Node) error {
		return n.Wait()
	})
}
func (et *ExecutingTask) GetOutput(name string) (Output, error) {
	if o, ok := et.outputs[name]; ok {
		return o, nil
	} else {
		return nil, fmt.Errorf("unknown output %s", name)
	}
}