feat: restart pipeline with recovery

docs/test-cases/pipeline-recovery.md

@@ -0,0 +1,279 @@
+<!-- markdownlint-disable MD013 -->
+# Test case for the pipeline recovery feature
+
+## Test Case 01: Recovery not triggered for fatal error - DLQ
+
+**Priority** (low/medium/high):
+
+**Description**:
+Recovery is not triggered when there is an error writing to a DLQ.
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+version: "2.2"
+pipelines:
+  - id: file-pipeline
+    status: running
+    name: file-pipeline
+    description: test pipeline
+    connectors:
+      - id: chaos-src
+        type: source
+        plugin: standalone:chaos
+        name: chaos-src
+        settings:
+          readMode: error
+      - id: log-dst
+        type: destination
+        plugin: builtin:log
+        log: file-dst
+    dead-letter-queue:
+      plugin: "builtin:postgres"
+      settings:
+        table: non_existing_table_so_that_dlq_fails
+        url: postgresql://meroxauser:meroxapass@localhost/meroxadb?sslmode=disable
+      window-size: 3
+      window-nack-threshold: 2
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 02: Recovery not triggered for fatal error - processor
+
+**Priority** (low/medium/high):
+
+**Description**:
+Recovery is not triggered when there is an error processing a record.
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 03: Recovery not triggered - graceful shutdown
+
+**Priority** (low/medium/high):
+
+**Description**:
+Recovery is not triggered when Conduit is shutting down gracefully (i.e. when
+typing Ctrl+C in the terminal where Conduit is running, or sending a SIGINT).
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 04: Recovery not triggered - user stopped pipeline
+
+**Priority** (low/medium/high):
+
+**Description**:
+Recovery is not triggered if a user stops a pipeline (via the HTTP API's
+`/v1/pipelines/pipeline-id/stop` endpoint).
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 05: Recovery is configured by default
+
+**Priority** (low/medium/high):
+
+**Description**:
+Pipeline recovery is configured by default. A failing pipeline will be restarted
+a number of times without any additional configuration.
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+version: "2.2"
+pipelines:
+    - id: chaos-to-log
+      status: running
+      name: chaos-to-log
+      description: chaos source, error on read
+      connectors:
+        - id: chaos-source-1
+          type: source
+          plugin: standalone:chaos
+          name: chaos-source-1
+          settings:
+            readMode: error
+        - id: destination1
+          type: destination
+          plugin: builtin:log
+          name: log-destination
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 06: Recovery not triggered on malformed pipeline
+
+**Priority** (low/medium/high):
+
+**Description**:
+Recovery is not triggered for a malformed pipeline, e.g. when a connector is
+missing.
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+version: "2.2"
+pipelines:
+    - id: nothing-to-log
+      status: running
+      name: nothing-to-log
+      description: no source
+      connectors:
+        - id: destination1
+          type: destination
+          plugin: builtin:log
+          name: log-destination
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 07: Conduit exits with --pipelines.exit-on-degraded=true and a pipeline failing after recovery
+
+**Priority** (low/medium/high):
+
+**Description**: Given a Conduit instance with
+`--pipelines.exit-on-degraded=true`, and a pipeline that's failing after the
+maximum number of retries configured, Conduit should shut down gracefully.
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 08: Conduit doesn't exit with --pipelines.exit-on-degraded=true and a pipeline that recovers after a few retries
+
+**Priority** (low/medium/high):
+
+**Description**:
+Given a Conduit instance with `--pipelines.exit-on-degraded=true`, and a
+pipeline that recovers after a few retries, Conduit should still be running.
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---
+
+## Test Case 09: Conduit exits with --pipelines.exit-on-degraded=true, --pipelines.error-recovery.max-retries=0, and a degraded pipeline
+
+**Priority** (low/medium/high):
+
+**Description**:
+Given a Conduit instance with
+`--pipelines.exit-on-degraded=true --pipelines.error-recovery.max-retries=0`,
+and a pipeline that goes into a degraded state, the Conduit instance will
+gracefully shut down. This is due `max-retries=0` disabling the recovery.
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+```
+
+**Steps**:
+
+**Expected Result**:
+
+**Additional comments**:
+
+---

docs/test-cases/test-case-template.md

@@ -0,0 +1,27 @@
+<!-- markdownlint-disable MD041 -->
+
+## Test Case 01: Test case title
+
+**Priority** (low/medium/high):
+
+**Description**:
+
+**Automated** (yes/no)
+
+**Setup**:
+
+**Pipeline configuration file**:
+
+```yaml
+```
+
+**Steps**:
+
+1. Step 1
+2. Step 2
+
+**Expected Result**:
+
+**Additional comments**:
+
+---

pkg/conduit/config.go

@@ -22,6 +22,7 @@ import (
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
 	"github.com/conduitio/conduit/pkg/foundation/log"
+	"github.com/conduitio/conduit/pkg/lifecycle"
 	"github.com/conduitio/conduit/pkg/plugin/connector/builtin"
 	"github.com/rs/zerolog"
 	"golang.org/x/exp/constraints"
@@ -84,14 +85,19 @@ type Config struct {
 	}
 
 	Pipelines struct {
-		Path          string
-		ExitOnError   bool
-		ErrorRecovery struct {
-			MinDelay      time.Duration
-			MaxDelay      time.Duration
+		Path           string
+		ExitOnDegraded bool
+		ErrorRecovery  struct {
+			// MinDelay is the minimum delay before restart: Default: 1 second
+			MinDelay time.Duration
+			// MaxDelay is the maximum delay before restart: Default: 10 minutes
+			MaxDelay time.Duration
+			// BackoffFactor is the factor by which the delay is multiplied after each restart: Default: 2
 			BackoffFactor int
-			MaxRetries    int
-			HealthyAfter  time.Duration
+			// MaxRetries is the maximum number of restarts before the pipeline is considered unhealthy: Default: -1 (infinite)
+			MaxRetries int64
+			// HealthyAfter is the time after which the pipeline is considered healthy: Default: 5 minutes
+			HealthyAfter time.Duration
 		}
 	}
 
@@ -137,7 +143,7 @@ func DefaultConfig() Config {
 	cfg.Pipelines.ErrorRecovery.MinDelay = time.Second
 	cfg.Pipelines.ErrorRecovery.MaxDelay = 10 * time.Minute
 	cfg.Pipelines.ErrorRecovery.BackoffFactor = 2
-	cfg.Pipelines.ErrorRecovery.MaxRetries = 0
+	cfg.Pipelines.ErrorRecovery.MaxRetries = lifecycle.InfiniteRetriesErrRecovery
 	cfg.Pipelines.ErrorRecovery.HealthyAfter = 5 * time.Minute
 
 	cfg.SchemaRegistry.Type = SchemaRegistryTypeBuiltin
@@ -206,8 +212,8 @@ func (c Config) validateErrorRecovery() error {
 	if err := requireNonNegativeValue("backoff-factor", errRecoveryCfg.BackoffFactor); err != nil {
 		errs = append(errs, err)
 	}
-	if err := requireNonNegativeValue("max-retries", errRecoveryCfg.MaxRetries); err != nil {
-		errs = append(errs, err)
+	if errRecoveryCfg.MaxRetries < lifecycle.InfiniteRetriesErrRecovery {
+		errs = append(errs, cerrors.Errorf(`invalid "max-retries" value. It must be %d for infinite retries or >= 0`, lifecycle.InfiniteRetriesErrRecovery))
 	}
 	if err := requirePositiveValue("healthy-after", errRecoveryCfg.HealthyAfter); err != nil {
 		errs = append(errs, err)

pkg/conduit/config_test.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/conduitio/conduit-commons/database/inmemory"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/lifecycle"
 	"github.com/matryer/is"
 )
 
@@ -196,12 +197,20 @@ func TestConfig_Validate(t *testing.T) {
 			want: cerrors.New(`invalid error recovery config: "backoff-factor" config value mustn't be negative (got: -1)`),
 		},
 		{
-			name: "error recovery: negative max-retries",
+			name: "error recovery: max-retries smaller than -1",
 			setupConfig: func(c Config) Config {
-				c.Pipelines.ErrorRecovery.MaxRetries = -1
+				c.Pipelines.ErrorRecovery.MaxRetries = lifecycle.InfiniteRetriesErrRecovery - 1
 				return c
 			},
-			want: cerrors.New(`invalid error recovery config: "max-retries" config value mustn't be negative (got: -1)`),
+			want: cerrors.New(`invalid error recovery config: invalid "max-retries" value. It must be -1 for infinite retries or >= 0`),
+		},
+		{
+			name: "error recovery: with 0 max-retries ",
+			setupConfig: func(c Config) Config {
+				c.Pipelines.ErrorRecovery.MaxRetries = 0
+				return c
+			},
+			want: nil,
 		},
 		{
 			name: "error recovery: negative healthy-after",