YQ Connector: prepare code base for S3 integration

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

6 files changed, 520 insertions, 0 deletions

diff --git a/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/client.go b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/client.go
new file mode 100644
index 0000000000..60b8298f86
--- /dev/null
+++ b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/client.go
@@ -0,0 +1,148 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/aws/middleware"
+	"github.com/aws/aws-sdk-go-v2/aws/retry"
+	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
+	"github.com/aws/smithy-go"
+	smithymiddleware "github.com/aws/smithy-go/middleware"
+	smithyhttp "github.com/aws/smithy-go/transport/http"
+)
+
+// ServiceID is the client identifer
+const ServiceID = "endpoint-credentials"
+
+// HTTPClient is a client for sending HTTP requests
+type HTTPClient interface {
+	Do(*http.Request) (*http.Response, error)
+}
+
+// Options is the endpoint client configurable options
+type Options struct {
+	// The endpoint to retrieve credentials from
+	Endpoint string
+
+	// The HTTP client to invoke API calls with. Defaults to client's default HTTP
+	// implementation if nil.
+	HTTPClient HTTPClient
+
+	// Retryer guides how HTTP requests should be retried in case of recoverable
+	// failures. When nil the API client will use a default retryer.
+	Retryer aws.Retryer
+
+	// Set of options to modify how the credentials operation is invoked.
+	APIOptions []func(*smithymiddleware.Stack) error
+}
+
+// Copy creates a copy of the API options.
+func (o Options) Copy() Options {
+	to := o
+	to.APIOptions = make([]func(*smithymiddleware.Stack) error, len(o.APIOptions))
+	copy(to.APIOptions, o.APIOptions)
+	return to
+}
+
+// Client is an client for retrieving AWS credentials from an endpoint
+type Client struct {
+	options Options
+}
+
+// New constructs a new Client from the given options
+func New(options Options, optFns ...func(*Options)) *Client {
+	options = options.Copy()
+
+	if options.HTTPClient == nil {
+		options.HTTPClient = awshttp.NewBuildableClient()
+	}
+
+	if options.Retryer == nil {
+		options.Retryer = retry.NewStandard()
+	}
+
+	for _, fn := range optFns {
+		fn(&options)
+	}
+
+	client := &Client{
+		options: options,
+	}
+
+	return client
+}
+
+// GetCredentialsInput is the input to send with the endpoint service to receive credentials.
+type GetCredentialsInput struct {
+	AuthorizationToken string
+}
+
+// GetCredentials retrieves credentials from credential endpoint
+func (c *Client) GetCredentials(ctx context.Context, params *GetCredentialsInput, optFns ...func(*Options)) (*GetCredentialsOutput, error) {
+	stack := smithymiddleware.NewStack("GetCredentials", smithyhttp.NewStackRequest)
+	options := c.options.Copy()
+	for _, fn := range optFns {
+		fn(&options)
+	}
+
+	stack.Serialize.Add(&serializeOpGetCredential{}, smithymiddleware.After)
+	stack.Build.Add(&buildEndpoint{Endpoint: options.Endpoint}, smithymiddleware.After)
+	stack.Deserialize.Add(&deserializeOpGetCredential{}, smithymiddleware.After)
+	retry.AddRetryMiddlewares(stack, retry.AddRetryMiddlewaresOptions{Retryer: options.Retryer})
+	middleware.AddSDKAgentKey(middleware.FeatureMetadata, ServiceID)
+	smithyhttp.AddErrorCloseResponseBodyMiddleware(stack)
+	smithyhttp.AddCloseResponseBodyMiddleware(stack)
+
+	for _, fn := range options.APIOptions {
+		if err := fn(stack); err != nil {
+			return nil, err
+		}
+	}
+
+	handler := smithymiddleware.DecorateHandler(smithyhttp.NewClientHandler(options.HTTPClient), stack)
+	result, _, err := handler.Handle(ctx, params)
+	if err != nil {
+		return nil, err
+	}
+
+	return result.(*GetCredentialsOutput), err
+}
+
+// GetCredentialsOutput is the response from the credential endpoint
+type GetCredentialsOutput struct {
+	Expiration      *time.Time
+	AccessKeyID     string
+	SecretAccessKey string
+	Token           string
+}
+
+// EndpointError is an error returned from the endpoint service
+type EndpointError struct {
+	Code    string            `json:"code"`
+	Message string            `json:"message"`
+	Fault   smithy.ErrorFault `json:"-"`
+}
+
+// Error is the error mesage string
+func (e *EndpointError) Error() string {
+	return fmt.Sprintf("%s: %s", e.Code, e.Message)
+}
+
+// ErrorCode is the error code returned by the endpoint
+func (e *EndpointError) ErrorCode() string {
+	return e.Code
+}
+
+// ErrorMessage is the error message returned by the endpoint
+func (e *EndpointError) ErrorMessage() string {
+	return e.Message
+}
+
+// ErrorFault indicates error fault classification
+func (e *EndpointError) ErrorFault() smithy.ErrorFault {
+	return e.Fault
+}
diff --git a/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/client_test.go b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/client_test.go
new file mode 100644
index 0000000000..ac7e37524b
--- /dev/null
+++ b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/client_test.go
@@ -0,0 +1,228 @@
+package client
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/smithy-go"
+	smithytesting "github.com/aws/smithy-go/testing"
+)
+
+func TestClient_GetCredentials(t *testing.T) {
+	cases := map[string]struct {
+		Token           string
+		RelativeURI     string
+		ResponseCode    int
+		ResponseBody    []byte
+		ExpectResult    *GetCredentialsOutput
+		ExpectErr       bool
+		ValidateRequest func(*testing.T, *http.Request)
+		ValidateError   func(*testing.T, error) bool
+	}{
+		"success static": {
+			ResponseCode: 200,
+			ResponseBody: []byte(` {
+   "AccessKeyId" : "FooKey",
+   "SecretAccessKey" : "FooSecret"
+ }`),
+			ExpectResult: &GetCredentialsOutput{
+				AccessKeyID:     "FooKey",
+				SecretAccessKey: "FooSecret",
+			},
+		},
+		"success with authorization token": {
+			Token:        "MySecretAuthToken",
+			ResponseCode: 200,
+			ResponseBody: []byte(` {
+   "AccessKeyId" : "FooKey",
+   "SecretAccessKey" : "FooSecret"
+ }`),
+			ExpectResult: &GetCredentialsOutput{
+				AccessKeyID:     "FooKey",
+				SecretAccessKey: "FooSecret",
+			},
+		},
+		"success refreshable": {
+			Token:        "MySecretAuthToken",
+			ResponseCode: 200,
+			ResponseBody: []byte(`{
+  "AccessKeyId": "FooKey",
+  "SecretAccessKey": "FooSecret",
+  "Token": "FooToken",
+  "Expiration": "2016-02-25T06:03:31Z"
+}`),
+			ExpectResult: &GetCredentialsOutput{
+				AccessKeyID:     "FooKey",
+				SecretAccessKey: "FooSecret",
+				Token:           "FooToken",
+				Expiration: func() *time.Time {
+					t := time.Date(2016, 2, 25, 6, 3, 31, 0, time.UTC)
+					return &t
+				}(),
+			},
+		},
+		"success with additional URI components": {
+			RelativeURI:  "/path/to/thing?someKey=someValue",
+			ResponseCode: 200,
+			ResponseBody: []byte(` {
+   "AccessKeyId" : "FooKey",
+   "SecretAccessKey" : "FooSecret"
+ }`),
+			ValidateRequest: func(t *testing.T, r *http.Request) {
+				t.Helper()
+				if e, a := "/path/to/thing", r.URL.Path; e != a {
+					t.Errorf("expect %v, got %v", e, a)
+				}
+				if e, a := "someValue", r.URL.Query().Get("someKey"); e != a {
+					t.Errorf("expect %v, got %v", e, a)
+				}
+			},
+			ExpectResult: &GetCredentialsOutput{
+				AccessKeyID:     "FooKey",
+				SecretAccessKey: "FooSecret",
+			},
+		},
+		"json error response client fault": {
+			ResponseCode: 403,
+			ResponseBody: []byte(`{
+  "code": "Unauthorized",
+  "message": "not authorized for endpoint"
+}`),
+			ExpectErr: true,
+			ValidateError: func(t *testing.T, err error) (ok bool) {
+				t.Helper()
+				var apiError smithy.APIError
+				if errors.As(err, &apiError) {
+					if e, a := "Unauthorized", apiError.ErrorCode(); e != a {
+						t.Errorf("expect %v, got %v", e, a)
+						ok = false
+					}
+					if e, a := "not authorized for endpoint", apiError.ErrorMessage(); e != a {
+						t.Errorf("expect %v, got %v", e, a)
+						ok = false
+					}
+					if e, a := smithy.FaultClient, apiError.ErrorFault(); e != a {
+						t.Errorf("expect %v, got %v", e, a)
+						ok = false
+					}
+				} else {
+					t.Errorf("expect %T error type, got %T: %v", apiError, err, err)
+					ok = false
+				}
+				return ok
+			},
+		},
+		"json error response server fault": {
+			ResponseCode: 500,
+			ResponseBody: []byte(`{
+  "code": "InternalError",
+  "message": "an error occurred"
+}`),
+			ExpectErr: true,
+			ValidateError: func(t *testing.T, err error) (ok bool) {
+				t.Helper()
+				var apiError smithy.APIError
+				if errors.As(err, &apiError) {
+					if e, a := "InternalError", apiError.ErrorCode(); e != a {
+						t.Errorf("expect %v, got %v", e, a)
+						ok = false
+					}
+					if e, a := "an error occurred", apiError.ErrorMessage(); e != a {
+						t.Errorf("expect %v, got %v", e, a)
+						ok = false
+					}
+					if e, a := smithy.FaultServer, apiError.ErrorFault(); e != a {
+						t.Errorf("expect %v, got %v", e, a)
+						ok = false
+					}
+				} else {
+					t.Errorf("expect %T error type, got %T: %v", apiError, err, err)
+					ok = false
+				}
+				return ok
+			},
+		},
+		"non-json error response": {
+			ResponseCode: 500,
+			ResponseBody: []byte(`<html><body>unexpected message format</body></html>`),
+			ExpectErr:    true,
+			ValidateError: func(t *testing.T, err error) (ok bool) {
+				t.Helper()
+				if e, a := "failed to decode error message", err.Error(); !strings.Contains(a, e) {
+					t.Errorf("expect %v, got %v", e, a)
+					ok = false
+				}
+				return ok
+			},
+		},
+	}
+
+	for name, tt := range cases {
+		t.Run(name, func(t *testing.T) {
+			var actualReq *http.Request
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				actualReq = r.Clone(r.Context())
+				var buf bytes.Buffer
+				if _, err := io.Copy(&buf, r.Body); err != nil {
+					t.Errorf("failed to read r body, %v", err)
+				}
+
+				actualReq.Body = ioutil.NopCloser(bytes.NewReader(buf.Bytes()))
+
+				w.WriteHeader(tt.ResponseCode)
+				w.Write(tt.ResponseBody)
+			}))
+			defer server.Close()
+
+			client := New(Options{Endpoint: server.URL + tt.RelativeURI})
+
+			result, err := client.GetCredentials(context.Background(), &GetCredentialsInput{AuthorizationToken: tt.Token})
+			if (err != nil) != tt.ExpectErr {
+				t.Fatalf("got error %v, but ExpectErr=%v", err, tt.ExpectErr)
+			}
+			if err != nil && tt.ValidateError != nil {
+				if !tt.ValidateError(t, err) {
+					return
+				}
+			}
+
+			if e, a := "application/json", actualReq.Header.Get("Accept"); e != a {
+				t.Errorf("expect %v, got %v", e, a)
+			}
+
+			if len(tt.Token) > 0 {
+				if e, a := tt.Token, actualReq.Header.Get("Authorization"); e != a {
+					t.Errorf("expect %v, got %v", e, a)
+				}
+			}
+
+			if tt.ValidateRequest != nil {
+				tt.ValidateRequest(t, actualReq)
+			}
+
+			if err = smithytesting.CompareValues(tt.ExpectResult, result); err != nil {
+				t.Errorf("expect value match:\n%v", err)
+			}
+		})
+	}
+}
+
+func TestClient_GetCredentials_NoEndpoint(t *testing.T) {
+	client := New(Options{})
+
+	_, err := client.GetCredentials(context.Background(), &GetCredentialsInput{})
+	if err == nil {
+		t.Fatalf("expect error got none")
+	}
+	if e, a := "endpoint not provided", err.Error(); !strings.Contains(a, e) {
+		t.Errorf("expect %v, got %v", e, a)
+	}
+}
diff --git a/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/gotest/ya.make b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/gotest/ya.make
new file mode 100644
index 0000000000..00dc081f11
--- /dev/null
+++ b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/gotest/ya.make
@@ -0,0 +1,5 @@
+GO_TEST_FOR(vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client)
+
+LICENSE(Apache-2.0)
+
+END()
diff --git a/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/middleware.go b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/middleware.go
new file mode 100644
index 0000000000..40747a53c1
--- /dev/null
+++ b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/middleware.go
@@ -0,0 +1,120 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/url"
+
+	"github.com/aws/smithy-go"
+	smithymiddleware "github.com/aws/smithy-go/middleware"
+	smithyhttp "github.com/aws/smithy-go/transport/http"
+)
+
+type buildEndpoint struct {
+	Endpoint string
+}
+
+func (b *buildEndpoint) ID() string {
+	return "BuildEndpoint"
+}
+
+func (b *buildEndpoint) HandleBuild(ctx context.Context, in smithymiddleware.BuildInput, next smithymiddleware.BuildHandler) (
+	out smithymiddleware.BuildOutput, metadata smithymiddleware.Metadata, err error,
+) {
+	request, ok := in.Request.(*smithyhttp.Request)
+	if !ok {
+		return out, metadata, fmt.Errorf("unknown transport, %T", in.Request)
+	}
+
+	if len(b.Endpoint) == 0 {
+		return out, metadata, fmt.Errorf("endpoint not provided")
+	}
+
+	parsed, err := url.Parse(b.Endpoint)
+	if err != nil {
+		return out, metadata, fmt.Errorf("failed to parse endpoint, %w", err)
+	}
+
+	request.URL = parsed
+
+	return next.HandleBuild(ctx, in)
+}
+
+type serializeOpGetCredential struct{}
+
+func (s *serializeOpGetCredential) ID() string {
+	return "OperationSerializer"
+}
+
+func (s *serializeOpGetCredential) HandleSerialize(ctx context.Context, in smithymiddleware.SerializeInput, next smithymiddleware.SerializeHandler) (
+	out smithymiddleware.SerializeOutput, metadata smithymiddleware.Metadata, err error,
+) {
+	request, ok := in.Request.(*smithyhttp.Request)
+	if !ok {
+		return out, metadata, fmt.Errorf("unknown transport type, %T", in.Request)
+	}
+
+	params, ok := in.Parameters.(*GetCredentialsInput)
+	if !ok {
+		return out, metadata, fmt.Errorf("unknown input parameters, %T", in.Parameters)
+	}
+
+	const acceptHeader = "Accept"
+	request.Header[acceptHeader] = append(request.Header[acceptHeader][:0], "application/json")
+
+	if len(params.AuthorizationToken) > 0 {
+		const authHeader = "Authorization"
+		request.Header[authHeader] = append(request.Header[authHeader][:0], params.AuthorizationToken)
+	}
+
+	return next.HandleSerialize(ctx, in)
+}
+
+type deserializeOpGetCredential struct{}
+
+func (d *deserializeOpGetCredential) ID() string {
+	return "OperationDeserializer"
+}
+
+func (d *deserializeOpGetCredential) HandleDeserialize(ctx context.Context, in smithymiddleware.DeserializeInput, next smithymiddleware.DeserializeHandler) (
+	out smithymiddleware.DeserializeOutput, metadata smithymiddleware.Metadata, err error,
+) {
+	out, metadata, err = next.HandleDeserialize(ctx, in)
+	if err != nil {
+		return out, metadata, err
+	}
+
+	response, ok := out.RawResponse.(*smithyhttp.Response)
+	if !ok {
+		return out, metadata, &smithy.DeserializationError{Err: fmt.Errorf("unknown transport type %T", out.RawResponse)}
+	}
+
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return out, metadata, deserializeError(response)
+	}
+
+	var shape *GetCredentialsOutput
+	if err = json.NewDecoder(response.Body).Decode(&shape); err != nil {
+		return out, metadata, &smithy.DeserializationError{Err: fmt.Errorf("failed to deserialize json response, %w", err)}
+	}
+
+	out.Result = shape
+	return out, metadata, err
+}
+
+func deserializeError(response *smithyhttp.Response) error {
+	var errShape *EndpointError
+	err := json.NewDecoder(response.Body).Decode(&errShape)
+	if err != nil {
+		return &smithy.DeserializationError{Err: fmt.Errorf("failed to decode error message, %w", err)}
+	}
+
+	if response.StatusCode >= 500 {
+		errShape.Fault = smithy.FaultServer
+	} else {
+		errShape.Fault = smithy.FaultClient
+	}
+
+	return errShape
+}
diff --git a/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/ya.make b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/ya.make
new file mode 100644
index 0000000000..19cb81ab0b
--- /dev/null
+++ b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/client/ya.make
@@ -0,0 +1,16 @@
+GO_LIBRARY()
+
+LICENSE(Apache-2.0)
+
+SRCS(
+    client.go
+    middleware.go
+)
+
+GO_TEST_SRCS(client_test.go)
+
+END()
+
+RECURSE(
+    gotest
+)
diff --git a/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/ya.make b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/ya.make
new file mode 100644
index 0000000000..54d997a3b4
--- /dev/null
+++ b/vendor/github.com/aws/aws-sdk-go-v2/credentials/endpointcreds/internal/ya.make
@@ -0,0 +1,3 @@
+RECURSE(
+    client
+)