// IngestGetPipeline returns pipelines based on ID.
349633
func (c *Client) IngestGetPipeline(ids ...string) *IngestGetPipelineService {
	return NewIngestGetPipelineService(c).Id(ids...)
}
407640
func (this *pipe) getPipe(isResolved bool) (func(v interface{}) *Future, *Promise) {
	if isResolved {
		return this.pipeDoneTask, this.pipePromise
	} else {
		return this.pipeFailTask, this.pipePromise
	}
}
1094518
func (s *QueryObjectsInput) SetPipelineId(v string) *QueryObjectsInput {
	s.PipelineId = &v
	return s
}
923463
func (s *Job) SetPipelineId(v string) *Job {
	s.PipelineId = &v
	return s
}
31328
func (m *metricExtractionSink) Ingest(span *ssf.SSFSpan) error {
	var metricsCount int
	defer func() {
		atomic.AddInt64(&m.metricsGenerated, int64(metricsCount))
		atomic.AddInt64(&m.spansProcessed, 1)
	}()
	metrics, err := samplers.ConvertMetrics(span)
	if err != nil {
		if _, ok := err.(samplers.InvalidMetrics); ok {
			m.log.WithError(err).
				Warn("Could not parse metrics from SSF Message")
			m.SendSample(ssf.Count("ssf.error_total", 1, map[string]string{
				"packet_type": "ssf_metric",
				"step":        "extract_metrics",
				"reason":      "invalid_metrics",
			}))
		} else {
			m.log.WithError(err).Error("Unexpected error extracting metrics from SSF Message")
			m.SendSample(ssf.Count("ssf.error_total", 1, map[string]string{
				"packet_type": "ssf_metric",
				"step":        "extract_metrics",
				"reason":      "unexpected_error",
				"error":       err.Error(),
			}))
			return err
		}
	}
	metricsCount += len(metrics)
	m.sendMetrics(metrics)

	if err := protocol.ValidateTrace(span); err != nil {
		return err
	}
	// If we made it here, we are dealing with a fully-fledged
	// trace span, not just a mere carrier for Samples:

	indicatorMetrics, err := samplers.ConvertIndicatorMetrics(span, m.indicatorSpanTimerName, m.objectiveSpanTimerName)
	if err != nil {
		m.log.WithError(err).
			WithField("span_name", span.Name).
			Warn("Couldn't extract indicator metrics for span")
		return err
	}
	metricsCount += len(indicatorMetrics)

	spanMetrics, err := samplers.ConvertSpanUniquenessMetrics(span, 0.01)
	if err != nil {
		m.log.WithError(err).
			WithField("span_name", span.Name).
			Warn("Couldn't extract uniqueness metrics for span")
		return err
	}
	metricsCount += len(spanMetrics)

	m.sendMetrics(append(indicatorMetrics, spanMetrics...))
	return nil
}
1310554
func (s *ReadPipelineOutput) SetPipeline(v *Pipeline) *ReadPipelineOutput {
	s.Pipeline = v
	return s
}
24066
func derivePipeline(f func(lines []string) <-chan string, g func(line string) <-chan int) func([]string) <-chan int {
	return func(a []string) <-chan int {
		b := f(a)
		return deriveJoin(deriveFmap(g, b))
	}
}
885750
func (c *Client) GoGetTimestamp() *PipelineRequest {
	pr := newPipelineRequest()
	c.requests <- pr
	return pr
}
965432
func (pcs *PipelineConfigsService) Get(ctx context.Context, name string) (p *Pipeline, resp *APIResponse, err error) {

	apiVersion, err := pcs.client.getAPIVersion(ctx, "admin/pipelines/:pipeline_name")
	if err != nil {
		return nil, nil, err
	}

	p = &Pipeline{}
	_, resp, err = pcs.client.getAction(ctx, &APIClientRequest{
		Path:         "admin/pipelines/" + name,
		APIVersion:   apiVersion,
		ResponseBody: p,
	})

	return
}
27327
func (k *KafkaSpanSink) Ingest(span *ssf.SSFSpan) error {
	samples := &ssf.Samples{}
	defer metrics.Report(k.traceClient, samples)
	// If we're sampling less than 100%, we should check whether a span should
	// be sampled:
	if k.sampleTag != "" || k.sampleThreshold < uint32(math.MaxUint32) {
		var hashKey uint32
		var sampleTagValue string

		if k.sampleTag == "" {
			// If we haven't set a sampleTag, we'll be hashing based on the traceID

			sampleTagValue = strconv.FormatInt(span.TraceId, 10)
		} else {
			// If we've set a sampleTag, we'll be hashing based off of that tag's value.

			var exists bool
			sampleTagValue, exists = span.Tags[k.sampleTag]
			if !exists {
				// If the span isn't tagged appropriately, we should drop it, regardless
				// of our sample rate.
				k.logger.Debug("Rejected span without appropriate tag")
				samples.Add(ssf.Count(sinks.MetricKeyTotalSpansDropped, 1, map[string]string{"sink": k.Name()}))
				return nil
			}
		}

		// Lifted from https://github.com/stathat/consistent/blob/75142be0209ec69bb014c7a1ac7d1a3c892c6424/consistent.go#L238-L245:
		// if the sample tag value that we're hashing is shorter than 64 bytes, we
		// need to pad it with zeroes for the crc32.ChecksumIEEE function.
		if len(sampleTagValue) < 64 {
			var scratch [64]byte
			copy(scratch[:], sampleTagValue)
			hashKey = crc32.ChecksumIEEE(scratch[:len(sampleTagValue)])
		} else {
			hashKey = crc32.ChecksumIEEE([]byte(sampleTagValue))
		}

		// Reject any spans whose hash keys end up greater than the threshold that
		// we previously computed.
		if hashKey > k.sampleThreshold {
			k.logger.WithField("traceId", span.TraceId).WithField("sampleTag", k.sampleTag).WithField("sampleTagValue", sampleTagValue).WithField("hashKey", hashKey).WithField("sampleThreshold", k.sampleThreshold).Debug("Rejected span based off of sampling rules")
			samples.Add(ssf.Count(sinks.MetricKeyTotalSpansSkipped, 1, map[string]string{"sink": k.Name()}))
			return nil
		}
	}
	var enc sarama.Encoder
	switch k.serializer {
	case "json":
		j, err := json.Marshal(span)
		if err != nil {
			k.logger.Error("Error marshalling span")
			samples.Add(ssf.Count("kafka.span_marshal_error_total", 1, nil))
			return err
		}
		enc = sarama.StringEncoder(j)
	case "protobuf":
		p, err := proto.Marshal(span)
		if err != nil {
			k.logger.Error("Error marshalling span")
			samples.Add(ssf.Count("kafka.span_marshal_error_total", 1, nil))
			return err
		}
		enc = sarama.ByteEncoder(p)
	default:
		return fmt.Errorf("Unknown serialization format for encoding Kafka message: %s", k.serializer)
	}

	message := &sarama.ProducerMessage{
		Topic: k.topic,
		Value: enc,
	}

	select {
	case k.producer.Input() <- message:
		atomic.AddInt64(&k.spansFlushed, 1)
		return nil
	case _ = <-time.After(IngestTimeout):
		return IngestTimeoutError
	}
}
1204464
func (s *DbSession) Pipe(pipeline interface{}, document Document) *mgo.Pipe {
	return s.collection(document.CollectionName()).Pipe(pipeline)
}
110206
func (s *Service) PipelineInfo(ctx context.Context, pipelineIdentity string) (*Pipeline, error) {
	var pipeline Pipeline
	return &pipeline, s.Get(ctx, &pipeline, fmt.Sprintf("/pipelines/%v", pipelineIdentity), nil, nil)
}
344013
func (c *boltConn) PreparePipeline(queries ...string) (PipelineStmt, error) {
	if c.statement != nil {
		return nil, errors.New("An open statement already exists")
	}
	if c.closed {
		return nil, errors.New("Connection already closed")
	}
	c.statement = newPipelineStmt(queries, c)
	return c.statement, nil
}
102683
func (t *RedconTransport) AppendEntriesPipeline(target string) (
	raft.AppendPipeline, error,
) {
	return nil, raft.ErrPipelineReplicationNotSupported
}
684716
func (s *CreateJobInput) SetPipelineId(v string) *CreateJobInput {
	s.PipelineId = &v
	return s
}
398557
func (p *Processor) IngestLine(s *Stream, line string) error {
	// Initialize our annotation stream.
	if err := p.initialize(); err != nil {
		log.WithError(err).Errorf(p.ctx, "Failed to initialize.")
		return err
	}

	a := extractAnnotation(line)
	if a != "" {
		log.Debugf(p.ctx, "Annotation: %q", a)
	}

	var step *annotation.Step
	if s.Annotate {
		if a != "" {
			// Append our annotation to the annotation state. This may cause our
			// annotation callbacks to be invoked.
			if err := p.astate.Append(a); err != nil {
				log.Fields{
					log.ErrorKey: err,
					"stream":     s.Name,
					"annotation": a,
				}.Errorf(p.ctx, "Failed to process annotation.")
			}
		}

		// Use the step handler for the current step.
		step = p.astate.CurrentStep()
	} else {
		// Not handling annotations. Use our root step handler.
		step = p.astate.RootStep()
	}

	h, err := p.getStepHandler(step, true)
	if err != nil {
		return err
	}

	// Determine our injected annotations.
	injectedAnnotations := h.flushInjectedAnnotations()

	// Emit the "all" link if configured (at most once).
	if lg := p.o.LinkGenerator; lg != nil && s.EmitAllLink {
		injectedAnnotations = append(injectedAnnotations,
			buildAliasAnnotation("all", "stdio", lg.GetLink("**/stdout", "**/stderr")))

		s.EmitAllLink = false
	}

	// Get our root log stream handler. As an optimization, if "step" is
	// the root step, then "h" is already the root handler, so we don't need
	// to duplicate the lookup.
	//
	// We only need the handler if we're going to emit annotations to the root
	// stream.
	var rootHandler *stepHandler
	if !s.StripAnnotations && (a != "" || len(injectedAnnotations) > 0) {
		if rootStep := p.astate.RootStep(); rootStep != step {
			rootHandler, err = p.getStepHandler(rootStep, true)
			if err != nil {
				return err
			}
		} else {
			rootHandler = h
		}
	}

	// Handle annotation line.
	if a != "" {
		// If we're teeing annotations, emit this annotation.
		//
		// Some annotations (notably, STEP_LOG_LINE) contain actual content instead
		// of structural build hinting. We don't want to tee those unless we're also
		// teeing text.
		if s.Tee != nil && p.o.TeeAnnotations && (p.o.TeeText || !isContentAnnotation(a)) {
			if err := writeTextLine(s.Tee, line); err != nil {
				log.WithError(err).Errorf(h, "Failed to tee annotation line.")
				return err
			}
		}

		// If we're not stripping annotations, emit this to the root handler.
		if !s.StripAnnotations {
			if err := rootHandler.writeBaseStream(s, line); err != nil {
				log.WithError(err).Errorf(p.ctx, "Failed to send line to LogDog.")
				return err
			}
		}
	}

	// Emit any injected annotations.
	for _, anno := range injectedAnnotations {
		// Teeing?
		if s.Tee != nil && p.o.TeeAnnotations {
			if err := writeTextLine(s.Tee, anno); err != nil {
				log.WithError(err).Errorf(h, "Failed to tee annotation line.")
				return err
			}
		}

		// Not stripping?
		if !s.StripAnnotations {
			if err := rootHandler.writeBaseStream(s, anno); err != nil {
				log.WithError(err).Errorf(h, "Failed to send injected annotation line to LogDog.")
				return err
			}
		}
	}

	// If we're stripping text, write a warning message noting that this stream
	// will not have text in it.
	if !p.o.TeeText && s.Tee != nil {
		if !h.textStrippedNote {
			err := writeTextLine(s.Tee, "This build is configured to send log data exclusively to LogDog. "+
				"Please use the LogDog link on the build page to view this log stream.")
			if err != nil {
				log.WithError(err).Errorf(h, "Failed to write text stripped notice.")
				return err
			}

			h.textStrippedNote = true
		}

		// Add links to specific log streams as they are generated.
		injectTextStreamLines := h.flushInjectedTextStreamLines()
		for _, line := range injectTextStreamLines {
			if err := writeTextLine(s.Tee, line); err != nil {
				log.WithError(err).Errorf(h, "Failed to inject text stream line: %s", line)
				return err
			}
		}
	}

	// If this is a text line, and we're teeing text, emit this line.
	if a == "" {
		if s.Tee != nil && p.o.TeeText {
			if err := writeTextLine(s.Tee, line); err != nil {
				log.WithError(err).Errorf(h, "Failed to tee text line.")
				return err
			}
		}

		// Write to our LogDog stream.
		if err := h.writeBaseStream(s, line); err != nil {
			log.WithError(err).Errorf(p.ctx, "Failed to send line to LogDog.")
			return err
		}
	}

	return err
}
1370869
func (s *WebhookDefinition) SetTargetPipeline(v string) *WebhookDefinition {
	s.TargetPipeline = &v
	return s
}