func (command HelloWorldResource) create(session *session.Session,
	logger *logrus.Logger) (map[string]interface{}, error) {
	logger.Info("create: Hello ", command.Message)
	return map[string]interface{}{
		"Resource": "Created message: " + command.Message,
	}, nil
}

func userDefinedCustomResourceForwarder(customResource *customResourceInfo,
	event *json.RawMessage,
	context *LambdaContext,
	w http.ResponseWriter,
	logger *logrus.Logger) {

	var rawProps map[string]interface{}
	json.Unmarshal([]byte(*event), &rawProps)

	var lambdaEvent cloudformationresources.CloudFormationLambdaEvent
	jsonErr := json.Unmarshal([]byte(*event), &lambdaEvent)
	if jsonErr != nil {
		logger.WithFields(logrus.Fields{
			"RawEvent":       rawProps,
			"UnmarshalError": jsonErr,
		}).Warn("Raw event data")
		http.Error(w, jsonErr.Error(), http.StatusInternalServerError)
	}

	logger.WithFields(logrus.Fields{
		"LambdaEvent": lambdaEvent,
	}).Debug("CloudFormation user resource lambda event")

	// Create the new request and send it off
	customResourceRequest := &cloudformationresources.UserFuncResourceRequest{}
	customResourceRequest.LambdaHandler = func(requestType string,
		stackID string,
		properties map[string]interface{},
		logger *logrus.Logger) (map[string]interface{}, error) {

		//  Descend to get the "UserProperties" field iff defined by the customResource
		var userProperties map[string]interface{}
		if _, exists := lambdaEvent.ResourceProperties["UserProperties"]; exists {
			childProps, ok := lambdaEvent.ResourceProperties["UserProperties"].(map[string]interface{})
			if !ok {
				return nil, fmt.Errorf("Failed to extract UserProperties from payload")
			}
			userProperties = childProps
		}
		return customResource.userFunction(requestType, stackID, userProperties, logger)
	}
	customResourceRequest.RequestType = lambdaEvent.RequestType
	customResourceRequest.ResponseURL = lambdaEvent.ResponseURL
	customResourceRequest.StackID = lambdaEvent.StackID
	customResourceRequest.RequestID = lambdaEvent.RequestID
	customResourceRequest.LogicalResourceID = lambdaEvent.LogicalResourceID
	customResourceRequest.PhysicalResourceID = lambdaEvent.PhysicalResourceID
	customResourceRequest.LogGroupName = context.LogGroupName
	customResourceRequest.LogStreamName = context.LogStreamName
	customResourceRequest.ResourceProperties = lambdaEvent.ResourceProperties
	if "" == customResourceRequest.PhysicalResourceID {
		customResourceRequest.PhysicalResourceID = fmt.Sprintf("LogStreamName: %s", context.LogStreamName)
	}
	requestErr := cloudformationresources.Run(customResourceRequest, logger)
	if requestErr != nil {
		http.Error(w, requestErr.Error(), http.StatusInternalServerError)
	} else {
		fmt.Fprint(w, "CustomResource handled: "+lambdaEvent.LogicalResourceID)
	}
}

func newInstanceCleaner(
	cs *compute.Service,
	log *logrus.Logger,
	rateLimiter ratelimit.RateLimiter,
	rateLimitMaxCalls uint64,
	rateLimitDuration time.Duration,
	cutoffTime time.Time,
	projectID string,
	filters []string,
	noop bool,
) *instanceCleaner {
	return &instanceCleaner{
		cs:  cs,
		log: log.WithField("component", "instance_cleaner"),

		projectID: projectID,
		filters:   filters,

		noop: noop,

		CutoffTime: cutoffTime,

		rateLimiter:       rateLimiter,
		rateLimitMaxCalls: rateLimitMaxCalls,
		rateLimitDuration: rateLimitDuration,
	}
}

// LogrusLogger is a middleware that will log each request recieved, along with
// some useful information, to the given logger.
func LogrusLogger(logger *logrus.Logger, h http.Handler) http.Handler {
	fn := func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		entry := logger.WithFields(logrus.Fields{
			"request": r.RequestURI,
			"method":  r.Method,
			"remote":  r.RemoteAddr,
		})

		if id := r.Header.Get(RequestIDKey); id != "" {
			entry = entry.WithField("request_id", id)
		}

		// Wrap the writer so we can track data information.
		neww := WrapWriter(w)

		// Dispatch to the underlying handler.
		entry.Info("started handling request")
		h.ServeHTTP(neww, r)

		// Log final information.
		entry.WithFields(logrus.Fields{
			"bytes_written": neww.BytesWritten(),
			"status":        neww.Status(),
			"text_status":   http.StatusText(neww.Status()),
			"took":          time.Since(start),
		}).Info("completed handling request")
	}

	return http.HandlerFunc(fn)
}

// Main is the top of the pile.  Start here.
func Main(log *logrus.Logger) {
	opts := NewOptions()
	if opts.Debug {
		log.Level = logrus.DebugLevel
	}

	if opts.FileStorePrefix == "" {
		opts.FileStorePrefix = "tmp"
	}

	server, err := NewServer(opts, log, nil)
	if err != nil {
		log.Fatal(err)
	}

	port := os.Getenv("PORT")
	if port == "" {
		port = "9839"
	}

	addr := fmt.Sprintf(":%s", port)
	log.WithFields(logrus.Fields{
		"addr": addr,
	}).Info("artifacts-service listening")

	server.Run(addr)
}

func annotateDiscoveryInfo(template *gocf.Template, logger *logrus.Logger) *gocf.Template {
	for eachResourceID, eachResource := range template.Resources {
		// Only apply this to lambda functions
		if eachResource.Properties.CfnResourceType() == "AWS::Lambda::Function" {

			// Update the metdata with a reference to the output of each
			// depended on item...
			for _, eachDependsKey := range eachResource.DependsOn {
				dependencyOutputs, _ := outputsForResource(template, eachDependsKey, logger)
				if nil != dependencyOutputs && len(dependencyOutputs) != 0 {
					logger.WithFields(logrus.Fields{
						"Resource":  eachDependsKey,
						"DependsOn": eachResource.DependsOn,
						"Outputs":   dependencyOutputs,
					}).Debug("Resource metadata")
					safeMetadataInsert(eachResource, eachDependsKey, dependencyOutputs)
				}
			}
			// Also include standard AWS outputs at a resource level if a lambda
			// needs to self-discover other resources
			safeMetadataInsert(eachResource, TagLogicalResourceID, gocf.String(eachResourceID))
			safeMetadataInsert(eachResource, TagStackRegion, gocf.Ref("AWS::Region"))
			safeMetadataInsert(eachResource, TagStackID, gocf.Ref("AWS::StackId"))
			safeMetadataInsert(eachResource, TagStackName, gocf.Ref("AWS::StackName"))
		}
	}
	return template
}

// NewServer creates a new *Server with a router and its routes registered
func NewServer(opts *Options, log *logrus.Logger, md metadata.LookupSaver) (*Server, error) {
	var err error

	log.Debug("creating new server")
	srv := &Server{
		opts: opts,
		log:  log,
	}

	srv.setupRouter()
	srv.setupNegroni()

	err = srv.getMd(md)
	if err != nil {
		return nil, err
	}

	err = srv.setupStorer()
	if err != nil {
		return nil, err
	}

	err = srv.setupAuther()
	if err != nil {
		return nil, err
	}

	return srv, nil
}

func validateSpartaPreconditions(lambdaAWSInfos []*LambdaAWSInfo, logger *logrus.Logger) error {
	var errorText []string
	collisionMemo := make(map[string]string, 0)

	// 1 - check for duplicate golang function references.
	for _, eachLambda := range lambdaAWSInfos {
		testName := eachLambda.lambdaFnName
		if _, exists := collisionMemo[testName]; !exists {
			collisionMemo[testName] = testName
			// We'll always find our own lambda
			duplicateCount := 0
			for _, eachCheckLambda := range lambdaAWSInfos {
				if testName == eachCheckLambda.lambdaFnName {
					duplicateCount++
				}
			}
			// We'll always find our own lambda
			if duplicateCount > 1 {
				logger.WithFields(logrus.Fields{
					"CollisionCount": duplicateCount,
					"Name":           testName,
				}).Error("Detected Sparta lambda function associated with multiple LambdaAWSInfo structs")
				errorText = append(errorText, fmt.Sprintf("Multiple definitions of lambda: %s", testName))
			}
		}
	}
	if len(errorText) != 0 {
		return errors.New(strings.Join(errorText[:], "\n"))
	}
	return nil
}

func NewMesosLib(master string, log *logrus.Logger, frameworkInfo *mesosproto.FrameworkInfo, ip string, port int) *MesosLib {
	m := &MesosLib{
		Log:           log,
		master:        master,
		frameworkInfo: frameworkInfo,
		ip:            ip,
		port:          port,
		Router:        mux.NewRouter(),
		events: events{
			mesosproto.Event_REGISTERED: make(chan *mesosproto.Event, 64),
			mesosproto.Event_OFFERS:     make(chan *mesosproto.Event, 64),
			mesosproto.Event_UPDATE:     make(chan *mesosproto.Event, 64),
		},
	}

	if m.ip == "" {
		name, err := os.Hostname()
		if err != nil {
			log.Fatalf("Failed to get hostname: %+v", err)
		}

		addrs, err := net.LookupHost(name)
		if err != nil {
			log.Fatalf("Failed to get address for hostname %q: %+v", name, err)
		}

		for _, addr := range addrs {
			if m.ip == "" || !strings.HasPrefix(addr, "127") {
				m.ip = addr
			}
		}
	}
	m.initAPI()
	return m
}

func SetupLogger(logger *logrus.Logger) {
	fmter := &prefixed.TextFormatter{}
	logger.Formatter = fmter
	logger.Out = os.Stdout
	// Only log the info severity or above.
	logger.Level = logrus.InfoLevel
}

func LogError(r *http.Request, err error, info string, logger *log.Logger) {
	logger.WithFields(log.Fields{
		"error":  err.Error(),
		"method": r.Method,
		"url":    r.URL.String(),
	}).Error(info)
}

func stackLambdaResources(serviceName string, cf *cloudformation.CloudFormation, logger *logrus.Logger) (provisionedResources, error) {

	resources := make(provisionedResources, 0)
	nextToken := ""
	for {
		params := &cloudformation.ListStackResourcesInput{
			StackName: aws.String(serviceName),
		}
		if "" != nextToken {
			params.NextToken = aws.String(nextToken)
		}
		resp, err := cf.ListStackResources(params)

		if err != nil {
			logger.Error(err.Error())
			return nil, err
		}
		for _, eachSummary := range resp.StackResourceSummaries {
			if *eachSummary.ResourceType == "AWS::Lambda::Function" {
				resources = append(resources, eachSummary)
			}
		}
		if nil != resp.NextToken {
			nextToken = *resp.NextToken
		} else {
			break
		}
	}
	return resources, nil
}

// StackExists returns whether the given stackName or stackID currently exists
func StackExists(stackNameOrID string, awsSession *session.Session, logger *logrus.Logger) (bool, error) {
	cf := cloudformation.New(awsSession)

	describeStacksInput := &cloudformation.DescribeStacksInput{
		StackName: aws.String(stackNameOrID),
	}
	describeStacksOutput, err := cf.DescribeStacks(describeStacksInput)
	logger.WithFields(logrus.Fields{
		"DescribeStackOutput": describeStacksOutput,
	}).Debug("DescribeStackOutput results")

	exists := false
	if err != nil {
		logger.WithFields(logrus.Fields{
			"DescribeStackOutputError": err,
		}).Debug("DescribeStackOutput")

		// If the stack doesn't exist, then no worries
		if strings.Contains(err.Error(), "does not exist") {
			exists = false
		} else {
			return false, err
		}
	} else {
		exists = true
	}
	return exists, nil
}

// NewWithNameAndLogger returns a new middleware handler with the specified name
// and logger
func NewWithNameAndLogger(name string, l *logrus.Logger) echo.MiddlewareFunc {
	return func(next echo.HandlerFunc) echo.HandlerFunc {
		return func(c *echo.Context) error {
			start := time.Now()

			entry := l.WithFields(logrus.Fields{
				"request": c.Request().RequestURI,
				"method":  c.Request().Method,
				"remote":  c.Request().RemoteAddr,
			})

			if reqID := c.Request().Header.Get("X-Request-Id"); reqID != "" {
				entry = entry.WithField("request_id", reqID)
			}

			entry.Info("started handling request")

			if err := next(c); err != nil {
				c.Error(err)
			}

			latency := time.Since(start)

			entry.WithFields(logrus.Fields{
				"status":      c.Response().Status(),
				"text_status": http.StatusText(c.Response().Status()),
				"took":        latency,
				fmt.Sprintf("measure#%s.latency", name): latency.Nanoseconds(),
			}).Info("completed handling request")

			return nil
		}
	}
}

func safeMergeTemplates(sourceTemplate *gocf.Template, destTemplate *gocf.Template, logger *logrus.Logger) error {
	var mergeErrors []string

	// Append the custom resources
	for eachKey, eachLambdaResource := range sourceTemplate.Resources {
		_, exists := destTemplate.Resources[eachKey]
		if exists {
			errorMsg := fmt.Sprintf("Duplicate CloudFormation resource name: %s", eachKey)
			mergeErrors = append(mergeErrors, errorMsg)
		} else {
			destTemplate.Resources[eachKey] = eachLambdaResource
		}
	}
	// Append the custom outputs
	for eachKey, eachLambdaOutput := range sourceTemplate.Outputs {
		_, exists := destTemplate.Outputs[eachKey]
		if exists {
			errorMsg := fmt.Sprintf("Duplicate CloudFormation output key name: %s", eachKey)
			mergeErrors = append(mergeErrors, errorMsg)
		} else {
			destTemplate.Outputs[eachKey] = eachLambdaOutput
		}
	}
	if len(mergeErrors) > 0 {
		logger.Error("Failed to update template. The following collisions were found:")
		for _, eachError := range mergeErrors {
			logger.Error("\t" + eachError)
		}
		return errors.New("Template merge failed")
	}
	return nil
}

func newImageCleaner(
	cs *compute.Service,
	log *logrus.Logger,
	rateLimiter ratelimit.RateLimiter,
	rateLimitMaxCalls uint64,
	rateLimitDuration time.Duration,
	projectID,
	jobBoardURL string,
	filters []string,
	noop bool,
) *imageCleaner {
	return &imageCleaner{
		cs:  cs,
		log: log.WithField("component", "image_cleaner"),

		projectID:   projectID,
		jobBoardURL: jobBoardURL,
		filters:     filters,

		noop: noop,

		rateLimiter:       rateLimiter,
		rateLimitMaxCalls: rateLimitMaxCalls,
		rateLimitDuration: rateLimitDuration,
	}
}

// LogRPCWithFields will feed any request context into a logrus Entry.
func LogRPCWithFields(log *logrus.Logger, ctx context.Context) *logrus.Entry {
	md, ok := metadata.FromContext(ctx)
	if !ok {
		return logrus.NewEntry(log)
	}
	return log.WithFields(MetadataToFields(md))
}

// Explore supports interactive command line invocation of the previously
// provisioned Sparta service
func Explore(lambdaAWSInfos []*LambdaAWSInfo, port int, logger *logrus.Logger) error {
	if 0 == port {
		port = 9999
	}
	urlHost := fmt.Sprintf("http://localhost:%d", port)
	logger.Info("The following URLs are available for testing.")

	msgText := ""

	// Get unique paths
	lambdaPaths := make(map[string]*LambdaAWSInfo, 0)
	for _, eachLambdaInfo := range lambdaAWSInfos {
		lambdaPaths[eachLambdaInfo.lambdaFnName] = eachLambdaInfo
	}

	for _, eachLambdaInfo := range lambdaPaths {
		functionPath := fmt.Sprintf("%s/%s", urlHost, eachLambdaInfo.lambdaFnName)
		logger.WithFields(logrus.Fields{
			"Path": functionPath,
		}).Info(eachLambdaInfo.lambdaFnName)

		if msgText == "" {
			msgText = fmt.Sprintf("\tcurl -vs -X POST -H \"Content-Type: application/json\" --data @testEvent.json %s", functionPath)
		}
	}
	logger.Info("Functions can be invoked via application/json over POST")
	logger.Info(msgText)
	logger.Info("Where @testEvent.json is a local file with top level `context` and `event` keys")
	// Start up the localhost server and publish the info
	return Execute(lambdaAWSInfos, port, 0, logger)
}

// Run manages invoking a user supplied function to perform
// the CloudFormation resource operation. Clients do not need
// to implement anything cloudformationresource related.
func Run(request *UserFuncResourceRequest, logger *logrus.Logger) error {
	logger.WithFields(logrus.Fields{
		"Name":    aws.SDKName,
		"Version": aws.SDKVersion,
	}).Debug("CloudFormation CustomResource AWS SDK info")

	operationOutputs, operationError := request.LambdaHandler(request.RequestType,
		request.StackID,
		request.ResourceProperties,
		logger)

	// Notify CloudFormation of the result
	if "" != request.ResponseURL {
		sendErr := sendCloudFormationResponse(&request.AbstractCustomResourceRequest,
			operationOutputs,
			operationError,
			logger)
		if nil != sendErr {
			logger.WithFields(logrus.Fields{
				"Error": sendErr.Error(),
			}).Info("Failed to notify CloudFormation of result.")
		} else {
			// If the cloudformation notification was complete, then this
			// execution functioned properly and we can clear the Error
			operationError = nil
		}
	}
	return operationError
}

// Execute creates an HTTP listener to dispatch execution. Typically
// called via Main() via command line arguments.
func Execute(lambdaAWSInfos []*LambdaAWSInfo, port int, parentProcessPID int, logger *logrus.Logger) error {
	if port <= 0 {
		port = defaultHTTPPort
	}
	logger.Info("Execute!")

	lookupMap := make(dispatchMap, 0)
	for _, eachLambdaInfo := range lambdaAWSInfos {
		lookupMap[eachLambdaInfo.lambdaFnName] = eachLambdaInfo
	}
	server := &http.Server{
		Addr:         fmt.Sprintf(":%d", port),
		Handler:      &lambdaHandler{lookupMap, logger},
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 10 * time.Second,
	}
	if 0 != parentProcessPID {
		logger.Debug("Sending SIGUSR2 to parent process: ", parentProcessPID)
		syscall.Kill(parentProcessPID, syscall.SIGUSR2)
	}
	logger.Debug("Binding to port: ", port)
	err := server.ListenAndServe()
	if err != nil {
		logger.Error("FAILURE: " + err.Error())
		return err
	}
	logger.Debug("Server available at: ", port)
	return nil
}