YQ Connector: prepare code base for S3 integration

1. Кодовая база Коннектора переписана с помощью Go дженериков так, чтобы добавление нового источника данных (в частности S3 + csv) максимально переиспользовало имеющийся код (чтобы сохранялась логика нарезания на блоки данных, учёт трафика и пр.)
2. API Connector расширено для работы с S3, но ещё пока не протестировано.

5 files changed, 267 insertions, 0 deletions

diff --git a/vendor/github.com/aws/smithy-go/waiter/gotest/ya.make b/vendor/github.com/aws/smithy-go/waiter/gotest/ya.make
new file mode 100644
index 0000000000..ff874c7e9b
--- /dev/null
+++ b/vendor/github.com/aws/smithy-go/waiter/gotest/ya.make
@@ -0,0 +1,5 @@
+GO_TEST_FOR(vendor/github.com/aws/smithy-go/waiter)
+
+LICENSE(Apache-2.0)
+
+END()
diff --git a/vendor/github.com/aws/smithy-go/waiter/logger.go b/vendor/github.com/aws/smithy-go/waiter/logger.go
new file mode 100644
index 0000000000..8d70a03ff2
--- /dev/null
+++ b/vendor/github.com/aws/smithy-go/waiter/logger.go
@@ -0,0 +1,36 @@
+package waiter
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/smithy-go/logging"
+	"github.com/aws/smithy-go/middleware"
+)
+
+// Logger is the Logger middleware used by the waiter to log an attempt
+type Logger struct {
+	// Attempt is the current attempt to be logged
+	Attempt int64
+}
+
+// ID representing the Logger middleware
+func (*Logger) ID() string {
+	return "WaiterLogger"
+}
+
+// HandleInitialize performs handling of request in initialize stack step
+func (m *Logger) HandleInitialize(ctx context.Context, in middleware.InitializeInput, next middleware.InitializeHandler) (
+	out middleware.InitializeOutput, metadata middleware.Metadata, err error,
+) {
+	logger := middleware.GetLogger(ctx)
+
+	logger.Logf(logging.Debug, fmt.Sprintf("attempting waiter request, attempt count: %d", m.Attempt))
+
+	return next.HandleInitialize(ctx, in)
+}
+
+// AddLogger is a helper util to add waiter logger after `SetLogger` middleware in
+func (m Logger) AddLogger(stack *middleware.Stack) error {
+	return stack.Initialize.Insert(&m, "SetLogger", middleware.After)
+}
diff --git a/vendor/github.com/aws/smithy-go/waiter/waiter.go b/vendor/github.com/aws/smithy-go/waiter/waiter.go
new file mode 100644
index 0000000000..03e46e2ee7
--- /dev/null
+++ b/vendor/github.com/aws/smithy-go/waiter/waiter.go
@@ -0,0 +1,66 @@
+package waiter
+
+import (
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/aws/smithy-go/rand"
+)
+
+// ComputeDelay computes delay between waiter attempts. The function takes in a current attempt count,
+// minimum delay, maximum delay, and remaining wait time for waiter as input. The inputs minDelay and maxDelay
+// must always be greater than 0, along with minDelay lesser than or equal to maxDelay.
+//
+// Returns the computed delay and if next attempt count is possible within the given input time constraints.
+// Note that the zeroth attempt results in no delay.
+func ComputeDelay(attempt int64, minDelay, maxDelay, remainingTime time.Duration) (delay time.Duration, err error) {
+	// zeroth attempt, no delay
+	if attempt <= 0 {
+		return 0, nil
+	}
+
+	// remainingTime is zero or less, no delay
+	if remainingTime <= 0 {
+		return 0, nil
+	}
+
+	// validate min delay is greater than 0
+	if minDelay == 0 {
+		return 0, fmt.Errorf("minDelay must be greater than zero when computing Delay")
+	}
+
+	// validate max delay is greater than 0
+	if maxDelay == 0 {
+		return 0, fmt.Errorf("maxDelay must be greater than zero when computing Delay")
+	}
+
+	// Get attempt ceiling to prevent integer overflow.
+	attemptCeiling := (math.Log(float64(maxDelay/minDelay)) / math.Log(2)) + 1
+
+	if attempt > int64(attemptCeiling) {
+		delay = maxDelay
+	} else {
+		// Compute exponential delay based on attempt.
+		ri := 1 << uint64(attempt-1)
+		// compute delay
+		delay = minDelay * time.Duration(ri)
+	}
+
+	if delay != minDelay {
+		// randomize to get jitter between min delay and delay value
+		d, err := rand.CryptoRandInt63n(int64(delay - minDelay))
+		if err != nil {
+			return 0, fmt.Errorf("error computing retry jitter, %w", err)
+		}
+
+		delay = time.Duration(d) + minDelay
+	}
+
+	// check if this is the last attempt possible and compute delay accordingly
+	if remainingTime-delay <= minDelay {
+		delay = remainingTime - minDelay
+	}
+
+	return delay, nil
+}
diff --git a/vendor/github.com/aws/smithy-go/waiter/waiter_test.go b/vendor/github.com/aws/smithy-go/waiter/waiter_test.go
new file mode 100644
index 0000000000..009b1ede67
--- /dev/null
+++ b/vendor/github.com/aws/smithy-go/waiter/waiter_test.go
@@ -0,0 +1,144 @@
+package waiter
+
+import (
+	mathrand "math/rand"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/smithy-go/rand"
+)
+
+func TestComputeDelay(t *testing.T) {
+	cases := map[string]struct {
+		totalAttempts       int64
+		minDelay            time.Duration
+		maxDelay            time.Duration
+		maxWaitTime         time.Duration
+		expectedMaxDelays   []time.Duration
+		expectedError       string
+		expectedMinAttempts int
+	}{
+		"standard": {
+			totalAttempts:       8,
+			minDelay:            2 * time.Second,
+			maxDelay:            120 * time.Second,
+			maxWaitTime:         300 * time.Second,
+			expectedMaxDelays:   []time.Duration{2, 4, 8, 16, 32, 64, 120, 120},
+			expectedMinAttempts: 8,
+		},
+		"zero minDelay": {
+			totalAttempts: 3,
+			minDelay:      0,
+			maxDelay:      120 * time.Second,
+			maxWaitTime:   300 * time.Second,
+			expectedError: "minDelay must be greater than zero",
+		},
+		"zero maxDelay": {
+			totalAttempts: 3,
+			minDelay:      10 * time.Second,
+			maxDelay:      0,
+			maxWaitTime:   300 * time.Second,
+			expectedError: "maxDelay must be greater than zero",
+		},
+		"zero remaining time": {
+			totalAttempts:       3,
+			minDelay:            10 * time.Second,
+			maxDelay:            20 * time.Second,
+			maxWaitTime:         0,
+			expectedMaxDelays:   []time.Duration{0},
+			expectedMinAttempts: 1,
+		},
+		"max wait time is less than min delay": {
+			totalAttempts:       3,
+			minDelay:            10 * time.Second,
+			maxDelay:            20 * time.Second,
+			maxWaitTime:         5 * time.Second,
+			expectedMaxDelays:   []time.Duration{0},
+			expectedMinAttempts: 1,
+		},
+		"large minDelay": {
+			totalAttempts:       80,
+			minDelay:            150 * time.Minute,
+			maxDelay:            200 * time.Minute,
+			maxWaitTime:         250 * time.Minute,
+			expectedMinAttempts: 1,
+		},
+		"large maxDelay": {
+			totalAttempts:       80,
+			minDelay:            15 * time.Minute,
+			maxDelay:            2000 * time.Minute,
+			maxWaitTime:         250 * time.Minute,
+			expectedMinAttempts: 5,
+		},
+	}
+
+	for name, c := range cases {
+		t.Run(name, func(t *testing.T) {
+			// mock smithy-go rand/#Reader
+			r := rand.Reader
+			defer func() {
+				rand.Reader = r
+			}()
+			rand.Reader = mathrand.New(mathrand.NewSource(1))
+
+			// mock waiter call
+			delays, err := mockwait(c.totalAttempts, c.minDelay, c.maxDelay, c.maxWaitTime)
+
+			if len(c.expectedError) != 0 {
+				if err == nil {
+					t.Fatalf("expected error, got none")
+				}
+				if e, a := c.expectedError, err.Error(); !strings.Contains(a, e) {
+					t.Fatalf("expected error %v, got %v instead", e, a)
+				}
+			} else if err != nil {
+				t.Fatalf("expected no error, got %v", err)
+			}
+
+			if e, a := c.expectedMinAttempts, len(delays); e > a {
+				t.Logf("%v", delays)
+				t.Fatalf("expected minimum attempts to be %v, got %v", e, a)
+			}
+
+			for i, expectedDelay := range c.expectedMaxDelays {
+				if e, a := expectedDelay*time.Second, delays[i]; e < a {
+					t.Fatalf("attempt %d : expected delay to be less than %v, got %v", i+1, e, a)
+				}
+
+				if e, a := c.minDelay, delays[i]; e > a && c.maxWaitTime > c.minDelay {
+					t.Fatalf("attempt %d : expected delay to be more than %v, got %v", i+1, e, a)
+				}
+			}
+			t.Logf("delays : %v", delays)
+		})
+	}
+}
+
+func mockwait(maxAttempts int64, minDelay, maxDelay, maxWaitTime time.Duration) ([]time.Duration, error) {
+	delays := make([]time.Duration, 0)
+	remainingTime := maxWaitTime
+	var attempt int64
+
+	for {
+		attempt++
+
+		if maxAttempts < attempt {
+			break
+		}
+
+		delay, err := ComputeDelay(attempt, minDelay, maxDelay, remainingTime)
+		if err != nil {
+			return delays, err
+		}
+
+		delays = append(delays, delay)
+
+		remainingTime -= delay
+		if remainingTime < minDelay {
+			break
+		}
+	}
+
+	return delays, nil
+}
diff --git a/vendor/github.com/aws/smithy-go/waiter/ya.make b/vendor/github.com/aws/smithy-go/waiter/ya.make
new file mode 100644
index 0000000000..b0eea7b4e8
--- /dev/null
+++ b/vendor/github.com/aws/smithy-go/waiter/ya.make
@@ -0,0 +1,16 @@
+GO_LIBRARY()
+
+LICENSE(Apache-2.0)
+
+SRCS(
+    logger.go
+    waiter.go
+)
+
+GO_TEST_SRCS(waiter_test.go)
+
+END()
+
+RECURSE(
+    gotest
+)