// DriftDetector is a detector that ensures that the service hasn't
// experienced configuration drift prior to being overwritten by a new provisioning
// step.
163698
func DriftDetector(errorOnDrift bool) sparta.ServiceValidationHookHandler {

	driftDetector := func(context map[string]interface{},
		serviceName string,
		template *gocf.Template,
		S3Bucket string,
		S3Key string,
		buildID string,
		awsSession *session.Session,
		noop bool,
		logger *logrus.Logger) error {
		// Create a cloudformation service.
		cfSvc := cloudformation.New(awsSession)
		detectStackDrift, detectStackDriftErr := cfSvc.DetectStackDrift(&cloudformation.DetectStackDriftInput{
			StackName: aws.String(serviceName),
		})
		if detectStackDriftErr != nil {
			// If it doesn't exist, then no worries...
			if strings.Contains(detectStackDriftErr.Error(), "does not exist") {
				return nil
			}
			return errors.Wrapf(detectStackDriftErr, "attempting to determine stack drift")
		}

		// Poll until it's done...
		describeDriftDetectionStatus := &cloudformation.DescribeStackDriftDetectionStatusInput{
			StackDriftDetectionId: detectStackDrift.StackDriftDetectionId,
		}
		detectionComplete := false

		// Put a limit on the detection
		for i := 0; i <= 30 && !detectionComplete; i++ {
			driftStatus, driftStatusErr := cfSvc.DescribeStackDriftDetectionStatus(describeDriftDetectionStatus)
			if driftStatusErr != nil {
				logger.WithField("error", driftStatusErr).Warn("Failed to check Stack Drift")
			}
			if driftStatus != nil {
				switch *driftStatus.DetectionStatus {
				case "DETECTION_COMPLETE":
					detectionComplete = true
				default:
					logger.WithField("Status", *driftStatus.DetectionStatus).
						Info("Waiting for drift detection to complete")
					time.Sleep(11 * time.Second)
				}
			}
		}
		if !detectionComplete {
			return errors.Errorf("Stack drift detection did not complete in time")
		}

		golangFuncName := func(logicalResourceID string) string {
			templateRes, templateResExists := template.Resources[logicalResourceID]
			if !templateResExists {
				return ""
			}
			metadata := templateRes.Metadata
			if len(metadata) <= 0 {
				metadata = make(map[string]interface{})
			}
			golangFunc, golangFuncExists := metadata["golangFunc"]
			if !golangFuncExists {
				return ""
			}
			switch typedFunc := golangFunc.(type) {
			case string:
				return typedFunc
			default:
				return fmt.Sprintf("%#v", typedFunc)
			}
		}

		// Log the drifts
		logDrifts := func(stackResourceDrifts []*cloudformation.StackResourceDrift) {
			for _, eachDrift := range stackResourceDrifts {
				if len(eachDrift.PropertyDifferences) != 0 {
					for _, eachDiff := range eachDrift.PropertyDifferences {
						entry := logger.WithFields(logrus.Fields{
							"Resource":       *eachDrift.LogicalResourceId,
							"Actual":         *eachDiff.ActualValue,
							"Expected":       *eachDiff.ExpectedValue,
							"Relation":       *eachDiff.DifferenceType,
							"PropertyPath":   *eachDiff.PropertyPath,
							"LambdaFuncName": golangFuncName(*eachDrift.LogicalResourceId),
						})
						if errorOnDrift {
							entry.Error("Stack drift detected")
						} else {
							entry.Warn("Stack drift detected")
						}
					}
				}
			}
		}

		// Utility function to fetch all the drifts
		stackResourceDrifts := make([]*cloudformation.StackResourceDrift, 0)
		input := &cloudformation.DescribeStackResourceDriftsInput{
			MaxResults: aws.Int64(100),
			StackName:  aws.String(serviceName),
		}
		// There can't be more than 200 resources in the template
		// https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cloudformation-limits.html
		loopCounter := 0
		for {
			driftResults, driftResultsErr := cfSvc.DescribeStackResourceDrifts(input)
			if driftResultsErr != nil {
				return errors.Wrapf(driftResultsErr, "attempting to describe stack drift")
			}
			stackResourceDrifts = append(stackResourceDrifts, driftResults.StackResourceDrifts...)
			if driftResults.NextToken == nil {
				break
			}
			loopCounter++
			// If there is more than 10 (1k total) something is seriously wrong...
			if loopCounter >= 10 {
				logDrifts(stackResourceDrifts)
				return errors.Errorf("Exceeded maximum number of Stack resource drifts: %d", len(stackResourceDrifts))
			}

			input = &cloudformation.DescribeStackResourceDriftsInput{
				MaxResults: aws.Int64(100),
				StackName:  aws.String(serviceName),
				NextToken:  driftResults.NextToken,
			}
		}

		// Log them
		logDrifts(stackResourceDrifts)
		if len(stackResourceDrifts) == 0 || !errorOnDrift {
			return nil
		}
		return errors.Errorf("stack %s operation prevented due to stack drift", serviceName)
	}
	return sparta.ServiceValidationHookFunc(driftDetector)
}
503732
func (t *Template) GetAllAWSGuardDutyDetectorResources() map[string]*resources.AWSGuardDutyDetector {
	results := map[string]*resources.AWSGuardDutyDetector{}
	for name, untyped := range t.Resources {
		switch resource := untyped.(type) {
		case *resources.AWSGuardDutyDetector:
			results[name] = resource
		}
	}
	return results
}
755510
func Tracer(sampleRate int, spanIDFunc, traceIDFunc IDFunc) goa.Middleware {
	return NewTracer(SamplingPercent(sampleRate), SpanIDFunc(spanIDFunc), TraceIDFunc(traceIDFunc))
}
361261
func PrefixDetector(prefix string, handler Handler) Detector {
	return Detector{
		Needed:  len([]byte(prefix)),
		Test:    func(b []byte) bool { return string(b) == prefix },
		Handler: handler,
	}
}
896517
func SubOptionMonitorUIDExtractor(extractor extractors.EventMetadataExtractor) UIDMonitorOption {
	return func(cfg *uidmonitor.Config) {
		cfg.EventMetadataExtractor = extractor
	}
}
458234
func (sd *defaultSlowDetector) Detect(eventCh <-chan *events.Envelope, errCh <-chan error) (<-chan *events.Envelope, <-chan error, slowDetectCh) {
	sd.logger.Println("[INFO] Start detecting slowConsumerAlert event")

	// Create new channel to pass producer
	eventCh_ := make(chan *events.Envelope)
	errCh_ := make(chan error)

	// doneCh is used to cancel sending data to
	// downstream process.
	sd.doneCh = make(chan struct{})

	// deteCh is used to send `slowConsumerAlert` event
	detectCh := make(slowDetectCh)

	// Detect from from trafficcontroller event messages
	go func() {
		defer close(eventCh_)
		for event := range eventCh {
			// Check nozzle can catch up firehose outputs speed.
			if isTruncated(event) {
				detectCh <- fmt.Errorf("doppler dropped messages from its queue because nozzle is slow")
			}

			select {
			case eventCh_ <- event:
			case <-sd.doneCh:
				// After doneCh is closed, sending event to downstream
				// is immediately stopped.
				return
			}

		}
	}()

	// Detect from websocket errors
	go func() {
		defer close(errCh_)
		for err := range errCh {
			switch t := err.(type) {
			case *websocket.CloseError:
				if t.Code == websocket.ClosePolicyViolation {
					// ClosePolicyViolation (1008)
					// indicates that an endpoint is terminating the connection
					// because it has received a message that violates its policy.
					//
					// This is a generic status code that can be returned when there is no
					// other more suitable status code (e.g., 1003 or 1009) or if there
					// is a need to hide specific details about the policy.
					//
					// http://tools.ietf.org/html/rfc6455#section-11.7
					detectCh <- fmt.Errorf(
						"websocket terminates the connection because connection is too slow (ClosePolicyViolation)")
				}
			}
			select {
			case errCh_ <- err:
			case <-sd.doneCh:
				// After doneCh is closed, sending events to downstream
				// is immediately stopped.
				return
			}

		}
	}()

	return eventCh_, errCh_, detectCh
}
493581
func (s *GetIPSetInput) SetDetectorId(v string) *GetIPSetInput {
	s.DetectorId = &v
	return s
}
885343
func (e SourceRepositoryEnumerator) Detect(dir string, noSourceDetection bool) (*SourceRepositoryInfo, error) {
	info := &SourceRepositoryInfo{
		Path: dir,
	}

	// no point in doing source-type detection if the requested build strategy
	// is docker or pipeline
	if !noSourceDetection {
		for _, d := range e.Detectors {
			if detected := d(dir); detected != nil {
				info.Types = append(info.Types, SourceLanguageType{
					Platform: detected.Platform,
					Version:  detected.Version,
				})
			}
		}
	}
	if path, ok, err := e.DockerfileTester.Has(dir); err == nil && ok {
		dockerfile, err := NewDockerfileFromFile(path)
		if err != nil {
			return nil, err
		}
		info.Dockerfile = dockerfile
	}
	if _, ok, err := e.JenkinsfileTester.Has(dir); err == nil && ok {
		info.Jenkinsfile = true
	}

	return info, nil
}
1358838
func NewServiceChangeTracker(makeServiceInfo makeServicePortFunc, isIPv6Mode *bool, recorder record.EventRecorder) *ServiceChangeTracker {
	return &ServiceChangeTracker{
		items:           make(map[types.NamespacedName]*serviceChange),
		makeServiceInfo: makeServiceInfo,
		isIPv6Mode:      isIPv6Mode,
		recorder:        recorder,
	}
}
643822
func PrefixBytesDetector(prefix []byte, handler Handler) Detector {
	return Detector{
		Needed: len(prefix),
		Test: func(b []byte) bool {
			for i, v := range prefix {
				if b[i] != v {
					return false
				}
			}
			return true
		},
		Handler: handler,
	}
}
1236939
func (s *GetFindingsInput) SetDetectorId(v string) *GetFindingsInput {
	s.DetectorId = &v
	return s
}
60361
func (s *DeleteIPSetInput) SetDetectorId(v string) *DeleteIPSetInput {
	s.DetectorId = &v
	return s
}
450259
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		var tracer = o.Tracer
		if tracer == nil {
			tracer = opentracing.GlobalTracer()
		}

		spanContext, err := tracer.Extract(
			opentracing.TextMap,
			opentracing.TextMapCarrier(m.Metadata))

		if err != nil && err != opentracing.ErrSpanContextNotFound {
			return next(ctx, m)
		}

		handlerSpan := tracer.StartSpan(
			opts.Name,
			consumerOption{clientContext: spanContext},
			pubsubTag,
		)
		defer handlerSpan.Finish()

		ctx = opentracing.ContextWithSpan(ctx, handlerSpan)
		err = next(ctx, m)

		if err != nil {
			handlerSpan.SetTag("error", "true")
			handlerSpan.SetTag("stack", err.Error())
		}

		return err
	}
}
505994
func (imu *IMUDriver) EnableShockDetection(detect bool) error {
	var d byte
	if detect {
		d = 1
	}
	return imu.connection.WriteSysex([]byte{CURIE_IMU, CURIE_IMU_SHOCK_DETECT, d})
}
97435
func NewSpdyStreamProvider(conn net.Conn, server bool) (StreamProvider, error) {
	spdyConn, spdyErr := spdystream.NewConnection(conn, server)
	if spdyErr != nil {
		return nil, spdyErr
	}
	provider := &spdyStreamProvider{
		conn:       spdyConn,
		closeChan:  make(chan struct{}),
		listenChan: make(chan *spdyStream),
	}
	go spdyConn.Serve(provider.newStreamHandler)

	return provider, nil
}
1014969
func (s *DeleteFilterInput) SetDetectorId(v string) *DeleteFilterInput {
	s.DetectorId = &v
	return s
}
1367264
func (s *ListIPSetsInput) SetDetectorId(v string) *ListIPSetsInput {
	s.DetectorId = &v
	return s
}