frostfs-http-gw/app.go

package main

import (
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"strconv"

	"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 {
		cli  sdk.Client
		pool pool.Client
		log  *zap.Logger
		cfg  *viper.Viper
		key  *ecdsa.PrivateKey

		wlog logger.Logger
		web  *fasthttp.Server

		jobDone chan struct{}
		webDone chan struct{}
	}

	App interface {
		Wait()
		Worker(context.Context)
		Serve(context.Context)
	}

	Option func(a *app)
)

func WithLogger(l *zap.Logger) Option {
	return func(a *app) {
		if l == nil {
			return
		}
		a.log = l
	}
}

func WithConfig(c *viper.Viper) Option {
	return func(a *app) {
		if c == nil {
			return
		}
		a.cfg = c
	}
}

func newApp(ctx context.Context, opt ...Option) App {
	a := &app{
		log: zap.L(),
		cfg: viper.GetViper(),
		web: new(fasthttp.Server),

		jobDone: make(chan struct{}),
		webDone: make(chan struct{}),
	}

	for i := range opt {
		opt[i](a)
	}

	a.wlog = logger.GRPC(a.log)

	if a.cfg.GetBool("verbose") {
		grpclog.SetLoggerV2(a.wlog)
	}

	conTimeout := a.cfg.GetDuration("connect_timeout")
	reqTimeout := a.cfg.GetDuration("request_timeout")
	tckTimeout := a.cfg.GetDuration("rebalance_timer")

	// -- setup FastHTTP server: --
	a.web.Name = "neofs-http-gate"
	a.web.ReadBufferSize = a.cfg.GetInt("web.read_buffer_size")
	a.web.WriteBufferSize = a.cfg.GetInt("web.write_buffer_size")
	a.web.ReadTimeout = a.cfg.GetDuration("web.read_timeout")
	a.web.WriteTimeout = a.cfg.GetDuration("web.write_timeout")
	a.web.GetOnly = true
	a.web.DisableHeaderNamesNormalizing = true
	a.web.NoDefaultServerHeader = true
	a.web.NoDefaultContentType = true
	// -- -- -- -- -- -- -- -- -- --

	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
		a.log.Info("add connection peer",
			zap.String("address", address),
			zap.Float64("weight", 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.WithTickerTimeout(tckTimeout),
		pool.WithConnectTimeout(conTimeout),
		pool.WithRequestTimeout(reqTimeout),
		pool.WithAPIPreparer(sdk.APIPreparer),
		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),
		sdk.WithAPIPreparer(sdk.APIPreparer))
	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 {
		log.Fatal("Don't use generated key, deprecated")

		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")

	select {
	case <-a.jobDone: // wait for job is stopped
		<-a.webDone
	case <-a.webDone: // wait for web-server is stopped
		<-a.jobDone
	}
}

func (a *app) Worker(ctx context.Context) {
	a.pool.Worker(ctx)
	close(a.jobDone)
}

func (a *app) Serve(ctx context.Context) {
	go func() {
		<-ctx.Done()
		a.log.Info("stop web-server", zap.Error(a.web.Shutdown()))
		close(a.webDone)
	}()

	r := router.New()
	r.RedirectTrailingSlash = true

	a.log.Info("enabled /get/{cid}/{oid}")
	r.GET("/get/{cid}/{oid}", a.receiveFile)

	// attaching /-/(ready,healthy)
	attachHealthy(r, a.pool.Status)

	// enable metrics
	if a.cfg.GetBool("metrics") {
		a.log.Info("enabled /metrics/")
		attachMetrics(r, a.wlog)
	}

	// enable pprof
	if a.cfg.GetBool("pprof") {
		a.log.Info("enabled /debug/pprof/")
		attachProfiler(r)
	}

	bind := a.cfg.GetString("listen_address")
	a.log.Info("run gateway server",
		zap.String("address", bind))

	a.web.Handler = r.Handler
	if err := a.web.ListenAndServe(bind); err != nil {
		a.log.Fatal("could not start server", zap.Error(err))
	}
}