Error running 1000s of tasks: "etcdserver: request is too large" #1186

.gitignore

@@ -8,3 +8,4 @@ cmd/**/debug
 hack/**/debug
 debug.test
 *.iml
+examples/k8s-jobs1.yaml

cmd/argo/commands/get.go

@@ -8,12 +8,13 @@ import (
  "strings"
  "text/tabwriter"
 
+ "github.com/argoproj/argo/errors"
+ wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+ "github.com/argoproj/argo/util/file"
  "github.com/argoproj/pkg/humanize"
  "github.com/ghodss/yaml"
  "github.com/spf13/cobra"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
- wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 )
 
 const onExitSuffix = "onExit"
@@ -36,6 +37,10 @@ func NewGetCommand() *cobra.Command {
  if err != nil {
  log.Fatal(err)
  }
+ err = CheckAndDecompress(wf)
+ if err != nil {
+ log.Fatal(err)
+ }
  printWorkflow(wf, output)
  },
  }
@@ -45,6 +50,21 @@ func NewGetCommand() *cobra.Command {
  return command
 }
 
+func CheckAndDecompress(wf *wfv1.Workflow) error {
+ if wf.Status.CompressedNodes != "" {
+ nodeContent, err := file.DecodeDecompressString(wf.Status.CompressedNodes)
+ if err != nil {
+ return errors.InternalWrapError(err)
+ }
+ err = json.Unmarshal([]byte(nodeContent), &wf.Status.Nodes)
+ if err != nil {
+ log.Fatal(err)
+ }
+ wf.Status.CompressedNodes = ""
+ }
+ return nil
+}
+
 func printWorkflow(wf *wfv1.Workflow, outFmt string) {
  switch outFmt {
  case "name":

cmd/argo/commands/list.go

@@ -134,6 +134,10 @@ func countPendingRunningCompleted(wf *wfv1.Workflow) (int, int, int) {
  pending := 0
  running := 0
  completed := 0
+ err := CheckAndDecompress(wf)
+ if err != nil {
+ log.Fatal(err)
+ }
  for _, node := range wf.Status.Nodes {
  tmpl := wf.GetTemplate(node.TemplateName)
  if tmpl == nil || !tmpl.IsPodType() {

cmd/argo/commands/logs.go

@@ -27,7 +27,7 @@ import (
 
  log "github.com/sirupsen/logrus"
  "github.com/spf13/cobra"
- "k8s.io/api/core/v1"
+ v1 "k8s.io/api/core/v1"
 )
 
 type logEntry struct {
@@ -136,6 +136,11 @@ func (p *logPrinter) PrintPodLogs(podName string) error {
 // Prints logs for workflow pod steps and return most recent log timestamp per pod name
 func (p *logPrinter) printRecentWorkflowLogs(wf *v1alpha1.Workflow) map[string]*time.Time {
  var podNodes []v1alpha1.NodeStatus
+ err := CheckAndDecompress(wf)
+ if err != nil {
+ log.Warn(err)
+ return nil
+ }
  for _, node := range wf.Status.Nodes {
  if node.Type == v1alpha1.NodeTypePod && node.Phase != v1alpha1.NodeError {
  podNodes = append(podNodes, node)
@@ -193,6 +198,11 @@ func (p *logPrinter) printLiveWorkflowLogs(workflowName string, wfClient workflo
  defer cancel()
 
  processPods := func(wf *v1alpha1.Workflow) {
+ err := CheckAndDecompress(wf)
+ if err != nil {
+ log.Warn(err)
+ return
+ }
  for id := range wf.Status.Nodes {
  node := wf.Status.Nodes[id]
  if node.Type == v1alpha1.NodeTypePod && node.Phase != v1alpha1.NodeError && streamedPods[node.ID] == false {

cmd/argo/commands/watch.go

@@ -45,6 +45,8 @@ func watchWorkflow(name string) {
  select {
  case next := <-watchIf.ResultChan():
  wf, _ = next.Object.(*wfv1.Workflow)
+ err := CheckAndDecompress(wf)
+ errors.CheckError(err)
  case <-ticker.C:
  }
  if wf == nil {

examples/compressed_workflow.yaml

@@ -0,0 +1,26 @@
+# Stress test to test upper bounds of concurrent pods
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+ generateName: pod-limits-
+spec:
+ entrypoint: pod-limits
+ parallelism: 5
+ arguments:
+ parameters:
+ - name: limit
+ value: 16
+
+ templates:
+ - name: pod-limits
+ steps:
+ - - name: run-pod
+ template: run-pod
+ withSequence:
+ count: "{{workflow.parameters.limit}}"
+
+ - name: run-pod
+ script:
+ image: alpine:latest
+ command: [sh]
+ source: head /dev/urandom | tr -dc A-Za-z0-9 | head -c 2048 ; echo ''

pkg/apis/workflow/v1alpha1/types.go

@@ -477,6 +477,9 @@ type WorkflowStatus struct {
  // A human readable message indicating details about why the workflow is in this condition.
  Message string `json:"message,omitempty"`
 
+ // Compressed and base64 decoded Nodes map
+ CompressedNodes string `json:"compressedNodes,omitempty"`
+
  // Nodes is a mapping between a node ID and the node's status.
  Nodes map[string]NodeStatus `json:"nodes,omitempty"`
 

util/file/fileutil.go

@@ -0,0 +1,97 @@
+package file
+
+import (
+ "archive/tar"
+ "bytes"
+ "compress/gzip"
+ "encoding/base64"
+ "io"
+ "io/ioutil"
+ "os"
+ "strings"
+
+ log "github.com/sirupsen/logrus"
+)
+
+// IsFileOrDirExistInGZip return true if file or directory exists in GZip file
+func IsFileOrDirExistInGZip(sourcePath string, gzipFilePath string) bool {
+
+ fi, err := os.Open(gzipFilePath)
+
+ if os.IsNotExist(err) {
+ return false
+ }
+ defer close(fi)
+
+ fz, err := gzip.NewReader(fi)
+ if err != nil {
+ return false
+ }
+ tr := tar.NewReader(fz)
+ for {
+ hdr, err := tr.Next()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+
+ return false
+ }
+ if hdr.FileInfo().IsDir() && strings.Contains(strings.Trim(hdr.Name, "/"), strings.Trim(sourcePath, "/")) {
+ return true
+ }
+ if strings.Contains(sourcePath, hdr.Name) && hdr.Size > 0 {
+ return true
+ }
+ }
+ return false
+}
+
+//Close the file
+func close(f io.Closer) {
+ err := f.Close()
+ if err != nil {
+ log.Warnf("Failed to close the file/writer/reader. %v", err)
+ }
+}
+
+// CompressEncodeString will return the compressed string with base64 encoded
+func CompressEncodeString(content string) string {
+ return base64.StdEncoding.EncodeToString(CompressContent([]byte(content)))
+}
+
+// DecodeDecompressString will return decode and decompress the
+func DecodeDecompressString(content string) (string, error) {
+
+ buf, err := base64.StdEncoding.DecodeString(content)
+ if err != nil {
+ return "", err
+ }
+ dBuf, err := DecompressContent(buf)
+ if err != nil {
+ return "", err
+ }
+ return string(dBuf), nil
+}
+
+// CompressContent will compress the byte array using zip writer
+func CompressContent(content []byte) []byte {
+ var buf bytes.Buffer
+ zipWriter := gzip.NewWriter(&buf)
+
+ _, err := zipWriter.Write(content)
+ if err != nil {
+ log.Warnf("Error in compressing: %v", err)
+ }
+ close(zipWriter)
+ return buf.Bytes()
+}
+
+// DecompressContent will return the uncompressed content
+func DecompressContent(content []byte) ([]byte, error) {
+
+ buf := bytes.NewReader(content)
+ gZipReader, _ := gzip.NewReader(buf)
+ defer close(gZipReader)
+ return ioutil.ReadAll(gZipReader)
+}

util/file/fileutil_test.go

@@ -0,0 +1,21 @@
+package file
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+// TestResubmitWorkflowWithOnExit ensures we do not carry over the onExit node even if successful
+func TestCompressContentString(t *testing.T) {
+ content := "{\"pod-limits-rrdm8-591645159\":{\"id\":\"pod-limits-rrdm8-591645159\",\"name\":\"pod-limits-rrdm8[0]." +
+ "run-pod(0:0)\",\"displayName\":\"run-pod(0:0)\",\"type\":\"Pod\",\"templateName\":\"run-pod\",\"phase\":" +
+ "\"Succeeded\",\"boundaryID\":\"pod-limits-rrdm8\",\"startedAt\":\"2019-03-07T19:14:50Z\",\"finishedAt\":" +
+ "\"2019-03-07T19:14:55Z\"}}"
+
+ compString := CompressEncodeString(content)
+
+ resultString, _ := DecodeDecompressString(compString)
+
+ assert.Equal(t, content, resultString)
+}

workflow/controller/controller.go

@@ -243,6 +243,16 @@ func (wfc *WorkflowController) processNextItem() bool {
  }
 
  woc := newWorkflowOperationCtx(wf, wfc)
+ //Decompress the node if it is compressed
+
+ err = woc.checkAndDecompress()
+ if err != nil {
+ log.Warnf("Failed to decompress '%s' to workflow object: %v", key, err)
+ woc.markWorkflowFailed(fmt.Sprintf("invalid spec: %s", err.Error()))
+ woc.persistUpdates()
+ wfc.throttler.Remove(key)
+ return true
+ }
  woc.operate()
  if woc.wf.Status.Completed() {
  wfc.throttler.Remove(key)