Migrate to CDN SDK

app.go

@@ -3,33 +3,39 @@ package main
 import (
 	"context"
 	"crypto/ecdsa"
-	"errors"
+	"crypto/elliptic"
+	"crypto/rand"
+	"strconv"
 	"time"
 
 	"github.com/fasthttp/router"
+	sdk "github.com/nspcc-dev/cdn-neofs-sdk"
+	"github.com/nspcc-dev/cdn-neofs-sdk/creds/neofs"
+	"github.com/nspcc-dev/cdn-neofs-sdk/logger"
+	"github.com/nspcc-dev/cdn-neofs-sdk/pool"
+	crypto "github.com/nspcc-dev/neofs-crypto"
 	"github.com/spf13/viper"
 	"github.com/valyala/fasthttp"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/grpclog"
+	"google.golang.org/grpc/keepalive"
 )
 
 type (
 	app struct {
-		pool *Pool
+		cli  sdk.Client
+		pool pool.Client
 		log  *zap.Logger
 		cfg  *viper.Viper
 		key  *ecdsa.PrivateKey
 
-		wlog logger
+		wlog logger.Logger
 		web  *fasthttp.Server
 
 		jobDone chan struct{}
 		webDone chan struct{}
 
-		rebalanceTimer time.Duration
-
-		nodes []string
-
 		conTimeout time.Duration
 		reqTimeout time.Duration
 	}
@@ -61,7 +67,7 @@ func WithConfig(c *viper.Viper) Option {
 	}
 }
 
-func newApp(opt ...Option) App {
+func newApp(ctx context.Context, opt ...Option) App {
 	a := &app{
 		log: zap.L(),
 		cfg: viper.GetViper(),
@@ -75,14 +81,12 @@ func newApp(opt ...Option) App {
 		opt[i](a)
 	}
 
-	a.wlog = gRPCLogger(a.log)
+	a.wlog = logger.GRPC(a.log)
 
 	if a.cfg.GetBool("verbose") {
 		grpclog.SetLoggerV2(a.wlog)
 	}
 
-	a.key = fetchKey(a.log, a.cfg)
-	a.rebalanceTimer = a.cfg.GetDuration("rebalance_timer")
 	a.conTimeout = a.cfg.GetDuration("connect_timeout")
 	a.reqTimeout = a.cfg.GetDuration("request_timeout")
 
@@ -98,11 +102,68 @@ func newApp(opt ...Option) App {
 	a.web.NoDefaultContentType = true
 	// -- -- -- -- -- -- -- -- -- --
 
-	a.pool = newPool(a.log, a.cfg, a.key)
+	connections := make(map[string]float64)
+	for i := 0; ; i++ {
+		address := a.cfg.GetString("peers." + strconv.Itoa(i) + ".address")
+		weight := a.cfg.GetFloat64("peers." + strconv.Itoa(i) + ".weight")
+		if address == "" {
+			break
+		}
+
+		connections[address] = weight
+	}
+
+	cred, err := prepareCredentials(a.cfg.GetString("key"), a.log)
+	if err != nil {
+		a.log.Fatal("could not prepare credentials", zap.Error(err))
+	}
+
+	a.pool, err = pool.New(ctx,
+		pool.WithLogger(a.log),
+		pool.WithCredentials(cred),
+		pool.WithWeightPool(connections),
+		pool.WithGRPCOptions(
+			grpc.WithBlock(),
+			grpc.WithInsecure(),
+			grpc.WithKeepaliveParams(keepalive.ClientParameters{
+				Time:                a.cfg.GetDuration("keepalive.time"),
+				Timeout:             a.cfg.GetDuration("keepalive.timeout"),
+				PermitWithoutStream: a.cfg.GetBool("keepalive.permit_without_stream"),
+			})))
+
+	if err != nil {
+		a.log.Fatal("could not prepare connection pool", zap.Error(err))
+	}
+
+	a.cli, err = sdk.New(ctx,
+		sdk.WithLogger(a.log),
+		sdk.WithCredentials(cred),
+		sdk.WithConnectionPool(a.pool))
+	if err != nil {
+		a.log.Fatal("could not prepare sdk client", zap.Error(err))
+	}
 
 	return a
 }
 
+func prepareCredentials(key string, log *zap.Logger) (neofs.Credentials, error) {
+	if key == generated {
+		sk, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+		if err != nil {
+			return nil, err
+		}
+
+		key, err = crypto.WIFEncode(sk)
+		if err != nil {
+			return nil, err
+		}
+
+		log.Info("generate new key", zap.String("wif", key))
+	}
+
+	return neofs.New(key)
+}
+
 func (a *app) Wait() {
 	a.log.Info("application started")
 
@@ -115,37 +176,7 @@ func (a *app) Wait() {
 }
 
 func (a *app) Worker(ctx context.Context) {
-	dur := a.rebalanceTimer
-	tick := time.NewTimer(dur)
-
-	a.pool.reBalance(ctx)
-
-	switch _, err := a.pool.getConnection(ctx); {
-	case err == nil:
-		// ignore
-	case errors.Is(err, context.Canceled):
-		// ignore
-		// l.Info("context canceled")
-	default:
-		a.log.Fatal("could get connection", zap.Error(err))
-	}
-
-loop:
-	for {
-		select {
-		case <-ctx.Done():
-			break loop
-		case <-tick.C:
-			a.pool.reBalance(ctx)
-			tick.Reset(dur)
-		}
-	}
-
-	a.pool.close()
-	tick.Stop()
-
-	a.log.Info("connection worker stopped")
-
+	a.pool.Worker(ctx)
 	close(a.jobDone)
 }
 
@@ -163,7 +194,7 @@ func (a *app) Serve(ctx context.Context) {
 	r.GET("/get/{cid}/{oid}", a.receiveFile)
 
 	// attaching /-/(ready,healthy)
-	attachHealthy(r, a.pool.unhealthy)
+	attachHealthy(r, a.pool.Status)
 
 	// enable metrics
 	if a.cfg.GetBool("metrics") {