Merge pull request #61 from s8sg/of-watchdog

Of watchdog

sdk/pipeline.go

@@ -23,9 +23,7 @@ type Pipeline struct {
 	ExecutionPosition map[string]string `json:"pipeline-execution-position"` // Denotes the node that is executing now
 	ExecutionDepth    int               `json:"pipeline-execution-depth"`    // Denotes the depth of subgraph its executing
 
-	CurrentDynamicOption   map[string]string   `json:"pipeline-dynamic-option"`           // Denotes the current dynamic option mapped against the dynamic Node UQ id
-	AllDynamicOption       map[string][]string `json:"pipeline-all-dynamic-options"`      // Denotes all options mapped  against the dynamic Node UQ id
-	DynamicDependencyCount map[string]int      `json:"pipeline-dynamic-dependency-count"` // Denotes the no of dependency for a nodes unique Id
+	CurrentDynamicOption map[string]string `json:"pipeline-dynamic-option"` // Denotes the current dynamic option mapped against the dynamic Node UQ id
 
 	FailureHandler PipelineErrorHandler `json:"-"`
 	Finally        PipelineHandler      `json:"-"`
@@ -37,12 +35,9 @@ func CreatePipeline() *Pipeline {
 	pipeline.Dag = NewDag()
 
 	pipeline.ExecutionPosition = make(map[string]string, 0)
-
+	pipeline.ExecutionDepth = 0
 	pipeline.CurrentDynamicOption = make(map[string]string, 0)
-	pipeline.AllDynamicOption = make(map[string][]string, 0)
-	pipeline.DynamicDependencyCount = make(map[string]int, 0)
 
-	pipeline.ExecutionDepth = 0
 	return pipeline
 }
 
@@ -67,24 +62,53 @@ func (pipeline *Pipeline) GetInitialNodeId() string {
 	return "0"
 }
 
+// GetNodeExecutionUniqueId provide a ID that is unique in an execution
+func (pipeline *Pipeline) GetNodeExecutionUniqueId(node *Node) string {
+	depth := 0
+	dag := pipeline.Dag
+	depthStr := ""
+	optionStr := ""
+	for depth < pipeline.ExecutionDepth {
+		depthStr = fmt.Sprintf("%d", depth)
+		node := dag.GetNode(pipeline.ExecutionPosition[depthStr])
+		option := pipeline.CurrentDynamicOption[node.GetUniqueId()]
+		if node.subDag != nil {
+			dag = node.subDag
+		} else {
+			dag = node.conditionalDags[option]
+		}
+		if optionStr == "" {
+			optionStr = option
+		} else {
+			optionStr = option + "--" + optionStr
+		}
+
+		depth++
+	}
+	if optionStr == "" {
+		return node.GetUniqueId()
+	}
+	return optionStr + "--" + node.GetUniqueId()
+}
+
 // GetCurrentNodeDag returns the current node and current dag based on execution position
 func (pipeline *Pipeline) GetCurrentNodeDag() (*Node, *Dag) {
-	index := 0
+	depth := 0
 	dag := pipeline.Dag
-	indexStr := ""
-	for index < pipeline.ExecutionDepth {
-		indexStr = fmt.Sprintf("%d", index)
-		node := dag.GetNode(pipeline.ExecutionPosition[indexStr])
+	depthStr := ""
+	for depth < pipeline.ExecutionDepth {
+		depthStr = fmt.Sprintf("%d", depth)
+		node := dag.GetNode(pipeline.ExecutionPosition[depthStr])
+		option := pipeline.CurrentDynamicOption[node.GetUniqueId()]
 		if node.subDag != nil {
 			dag = node.subDag
 		} else {
-			option := pipeline.CurrentDynamicOption[node.GetUniqueId()]
 			dag = node.conditionalDags[option]
 		}
-		index++
+		depth++
 	}
-	indexStr = fmt.Sprintf("%d", index)
-	node := dag.GetNode(pipeline.ExecutionPosition[indexStr])
+	depthStr = fmt.Sprintf("%d", depth)
+	node := dag.GetNode(pipeline.ExecutionPosition[depthStr])
 	return node, dag
 }
 
@@ -122,8 +146,5 @@ func (pipeline *Pipeline) ApplyState(state string) {
 	temp, _ := decodePipeline([]byte(state))
 	pipeline.ExecutionDepth = temp.ExecutionDepth
 	pipeline.ExecutionPosition = temp.ExecutionPosition
-
 	pipeline.CurrentDynamicOption = temp.CurrentDynamicOption
-	pipeline.AllDynamicOption = temp.AllDynamicOption
-	pipeline.DynamicDependencyCount = temp.DynamicDependencyCount
 }

template/faas-flow/Dockerfile

@@ -1,11 +1,10 @@
-FROM golang:1.10.5-alpine3.7 as builder
+FROM openfaas/of-watchdog:0.5.3 as watchdog
+FROM golang:1.10.4-alpine3.8 as build
 
-RUN apk --no-cache add curl \
-    && echo "Pulling watchdog binary from Github." \
-    && curl -sSL https://github.com/openfaas/faas/releases/download/0.8.0/fwatchdog > /usr/bin/fwatchdog \
-    && chmod +x /usr/bin/fwatchdog \
-    && apk del curl --no-cache
+COPY --from=watchdog /fwatchdog /usr/bin/fwatchdog
+RUN chmod +x /usr/bin/fwatchdog
 
+RUN mkdir -p /go/src/handler
 WORKDIR /go/src/handler
 COPY . .
 
@@ -16,26 +15,25 @@ RUN CGO_ENABLED=0 GOOS=linux \
     go build --ldflags "-s -w" -a -installsuffix cgo -o handler . && \
     go test $(go list ./... | grep -v /vendor/) -cover
 
-FROM alpine:3.7
-RUN apk --no-cache add ca-certificates
-
-# Add non root user
-RUN addgroup -S app && adduser -S -g app app
-RUN mkdir -p /home/app
+FROM alpine:3.8
+# Add non root user and certs
+RUN apk --no-cache add ca-certificates \
+    && addgroup -S app && adduser -S -g app app \
+    && mkdir -p /home/app \
+    && chown app /home/app
 
 WORKDIR /home/app
 
-COPY --from=builder /usr/bin/fwatchdog         .
-
-COPY --from=builder /go/src/handler/function/  .
-COPY --from=builder /go/src/handler/handler    .
+COPY --from=build /go/src/handler/handler    .
+COPY --from=build /usr/bin/fwatchdog         .
+COPY --from=build /go/src/handler/function/  .
 
-RUN chown app /home/app
+RUN chown -R app /home/app
 
 USER app
 
 ENV fprocess="./handler"
-
-HEALTHCHECK --interval=2s CMD [ -e /tmp/.lock ] || exit 1
+ENV mode="http"
+ENV upstream_url="http://127.0.0.1:8082"
 
 CMD ["./fwatchdog"]