reinit ingestion client on perm denied (#2)

Author: candiduslynx

client.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+	"unsafe"
+
+	"google.golang.org/grpc"
+
+	"github.com/yandex-cloud/go-genproto/yandex/cloud/logging/v1"
+	ycsdk "github.com/yandex-cloud/go-sdk"
+)
+
+type client struct {
+	mu     sync.RWMutex
+	writer logging.LogIngestionServiceClient
+	init   func() error
+}
+
+var _ logging.LogIngestionServiceClient = (*client)(nil)
+
+func (c *client) Write(ctx context.Context, in *logging.WriteRequest, opts ...grpc.CallOption) (*logging.WriteResponse, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.writer.Write(ctx, in, opts...)
+}
+
+var (
+	PluginVersion    string
+	FluentBitVersion string
+)
+
+func clientInit(c *client, plugin unsafe.Pointer) func() error {
+	var initTime time.Time
+	return func() error {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+
+		const initBackoff = 30 * time.Second
+		passed := time.Since(initTime)
+		if passed < initBackoff {
+			return fmt.Errorf("%s since last client init haven't passed, only %s", initBackoff, passed)
+		}
+
+		const (
+			keyAuthorization = "authorization"
+			keyEndpoint      = "endpoint"
+			defaultEndpoint  = "api.cloud.yandex.net:443"
+		)
+
+		authorization := getConfigKey(plugin, keyAuthorization)
+		if authorization == "" {
+			return fmt.Errorf("authorization missing")
+		}
+
+		credentials, err := makeCredentials(authorization)
+		if err != nil {
+			return err
+		}
+
+		endpoint := getConfigKey(plugin, keyEndpoint)
+		if endpoint == "" {
+			endpoint = defaultEndpoint
+		}
+
+		tlsConfig, err := makeTLSConfig(plugin)
+		if err != nil {
+			return fmt.Errorf("error creating tls config: %s", err.Error())
+		}
+
+		sdk, err := ycsdk.Build(context.Background(),
+			ycsdk.Config{
+				Credentials: credentials,
+				Endpoint:    endpoint,
+				TLSConfig:   tlsConfig,
+			},
+			grpc.WithUserAgent(`fluent-bit-plugin-yandex/`+PluginVersion+`; fluent-bit/`+FluentBitVersion),
+		)
+		if err != nil {
+			return fmt.Errorf("error creating sdk: %s", err.Error())
+		}
+		c.writer = sdk.LogIngestion().LogIngestion()
+		initTime = time.Now()
+		return nil
+	}
+}
+
+func getIngestionClient(plugin unsafe.Pointer) (*client, error) {
+	c := new(client)
+	c.init = clientInit(c, plugin)
+	return c, c.init()
+}

config.go

@@ -1,17 +1,12 @@
 package main
 
 import (
-	"context"
 	"fmt"
 	"strings"
 	"unsafe"
 
 	"github.com/fluent/fluent-bit-go/output"
-	"google.golang.org/grpc"
-
 	"github.com/yandex-cloud/go-genproto/yandex/cloud/logging/v1"
-	ycsdk "github.com/yandex-cloud/go-sdk"
-	ingest "github.com/yandex-cloud/go-sdk/gen/logingestion"
 )
 
 func getConfigKey(plugin unsafe.Pointer, key string) string {
@@ -104,49 +99,3 @@ func getParseKeys(plugin unsafe.Pointer) *parseKeys {
 		messageTag: getConfigKey(plugin, keyMessageTagKey),
 	}
 }
-
-var (
-	PluginVersion    string
-	FluentBitVersion string
-)
-
-func getIngestionClient(plugin unsafe.Pointer) (*ingest.LogIngestionServiceClient, error) {
-	const (
-		keyAuthorization = "authorization"
-		keyEndpoint      = "endpoint"
-		defaultEndpoint  = "api.cloud.yandex.net:443"
-	)
-
-	authorization := getConfigKey(plugin, keyAuthorization)
-	if authorization == "" {
-		return nil, fmt.Errorf("authorization missing")
-	}
-
-	credentials, err := makeCredentials(authorization)
-	if err != nil {
-		return nil, err
-	}
-
-	endpoint := getConfigKey(plugin, keyEndpoint)
-	if endpoint == "" {
-		endpoint = defaultEndpoint
-	}
-
-	tlsConfig, err := makeTLSConfig(plugin)
-	if err != nil {
-		return nil, fmt.Errorf("error creating tls config: %s", err.Error())
-	}
-
-	sdk, err := ycsdk.Build(context.Background(),
-		ycsdk.Config{
-			Credentials: credentials,
-			Endpoint:    endpoint,
-			TLSConfig:   tlsConfig,
-		},
-		grpc.WithUserAgent(`fluent-bit-plugin-yandex/`+PluginVersion+`; fluent-bit/`+FluentBitVersion),
-	)
-	if err != nil {
-		return nil, fmt.Errorf("error creating sdk: %s", err.Error())
-	}
-	return sdk.LogIngestion().LogIngestion(), nil
-}

go.mod

@@ -1,11 +1,25 @@
 module github.com/yandex-cloud/fluent-bit-plugin-yandex
 
-go 1.16
+go 1.17
 
 require (
 	github.com/fluent/fluent-bit-go v0.0.0-20201210173045-3fd1e0486df2
-	github.com/yandex-cloud/go-genproto v0.0.0-20210824140926-1bca7bc0c005
-	github.com/yandex-cloud/go-sdk v0.0.0-20210824141121-182aedd44a25
-	google.golang.org/grpc v1.28.0
-	google.golang.org/protobuf v1.26.0
+	github.com/yandex-cloud/go-genproto v0.0.0-20211202135052-789603780fb2
+	github.com/yandex-cloud/go-sdk v0.0.0-20211206101223-7c4e7926bf53
+	google.golang.org/grpc v1.42.0
+	google.golang.org/protobuf v1.27.1
+)
+
+require (
+	github.com/ghodss/yaml v1.0.0 // indirect
+	github.com/golang-jwt/jwt/v4 v4.2.0 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/ugorji/go/codec v1.1.7 // indirect
+	golang.org/x/net v0.0.0-20211205041911-012df41ee64c // indirect
+	golang.org/x/sys v0.0.0-20211205182925-97ca703d548d // indirect
+	golang.org/x/text v0.3.7 // indirect
+	google.golang.org/genproto v0.0.0-20211203200212-54befc351ae9 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 )