frostfs-http-gw/app.go

206 lines
4.6 KiB
Go
Raw Normal View History

2020-03-31 08:37:10 +00:00
package main
import (
"context"
"crypto/ecdsa"
2020-11-09 13:43:23 +00:00
"strconv"
2020-03-31 08:37:10 +00:00
"github.com/fasthttp/router"
sdk "github.com/nspcc-dev/cdn-sdk"
"github.com/nspcc-dev/cdn-sdk/creds/neofs"
"github.com/nspcc-dev/cdn-sdk/logger"
"github.com/nspcc-dev/cdn-sdk/pool"
2020-03-31 08:37:10 +00:00
"github.com/spf13/viper"
"github.com/valyala/fasthttp"
"go.uber.org/zap"
2020-11-09 13:43:23 +00:00
"google.golang.org/grpc"
2020-07-02 08:34:54 +00:00
"google.golang.org/grpc/grpclog"
2020-11-09 13:43:23 +00:00
"google.golang.org/grpc/keepalive"
2020-03-31 08:37:10 +00:00
)
type (
app struct {
2020-11-09 13:43:23 +00:00
cli sdk.Client
pool pool.Client
2020-03-31 08:37:10 +00:00
log *zap.Logger
cfg *viper.Viper
key *ecdsa.PrivateKey
2020-11-09 13:43:23 +00:00
wlog logger.Logger
2020-03-31 08:37:10 +00:00
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
}
}
2020-11-09 13:43:23 +00:00
func newApp(ctx context.Context, opt ...Option) App {
2020-03-31 08:37:10 +00:00
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)
}
2020-11-09 13:43:23 +00:00
a.wlog = logger.GRPC(a.log)
2020-03-31 08:37:10 +00:00
if a.cfg.GetBool(cmdVerbose) {
2020-03-31 08:37:10 +00:00
grpclog.SetLoggerV2(a.wlog)
}
conTimeout := a.cfg.GetDuration(cfgConTimeout)
reqTimeout := a.cfg.GetDuration(cfgReqTimeout)
tckTimeout := a.cfg.GetDuration(cfgRebalance)
2020-03-31 08:37:10 +00:00
// -- setup FastHTTP server: --
a.web.Name = "neofs-http-gate"
a.web.ReadBufferSize = a.cfg.GetInt(cfgWebReadBufferSize)
a.web.WriteBufferSize = a.cfg.GetInt(cfgWebWriteBufferSize)
a.web.ReadTimeout = a.cfg.GetDuration(cfgWebReadTimeout)
a.web.WriteTimeout = a.cfg.GetDuration(cfgWebWriteTimeout)
2020-03-31 08:37:10 +00:00
a.web.GetOnly = true
a.web.DisableHeaderNamesNormalizing = true
a.web.NoDefaultServerHeader = true
a.web.NoDefaultContentType = true
// -- -- -- -- -- -- -- -- -- --
2020-11-09 13:43:23 +00:00
connections := make(map[string]float64)
for i := 0; ; i++ {
address := a.cfg.GetString(cfgPeers + "." + strconv.Itoa(i) + ".address")
weight := a.cfg.GetFloat64(cfgPeers + "." + strconv.Itoa(i) + ".weight")
2020-11-09 13:43:23 +00:00
if address == "" {
break
}
connections[address] = weight
a.log.Info("add connection peer",
zap.String("address", address),
zap.Float64("weight", weight))
2020-11-09 13:43:23 +00:00
}
cred, err := neofs.New(a.cfg.GetString(cmdNeoFSKey))
2020-11-09 13:43:23 +00:00
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),
2020-11-09 13:43:23 +00:00
pool.WithGRPCOptions(
grpc.WithBlock(),
grpc.WithInsecure(),
grpc.WithKeepaliveParams(keepalive.ClientParameters{
Time: a.cfg.GetDuration(cfgKeepaliveTime),
Timeout: a.cfg.GetDuration(cfgKeepaliveTimeout),
PermitWithoutStream: a.cfg.GetBool(cfgKeepalivePermitWithoutStream),
2020-11-09 13:43:23 +00:00
})))
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))
2020-11-09 13:43:23 +00:00
if err != nil {
a.log.Fatal("could not prepare sdk client", zap.Error(err))
}
2020-03-31 08:37:10 +00:00
return a
}
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) {
2020-11-09 13:43:23 +00:00
a.pool.Worker(ctx)
2020-03-31 08:37:10 +00:00
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
2020-05-12 08:18:23 +00:00
a.log.Info("enabled /get/{cid}/{oid}")
r.GET("/get/{cid}/{oid}", a.byAddress)
a.log.Info("enabled /get_by_attribute/{cid}/{attr_key}/{attr_val}")
r.GET("/get_by_attribute/{cid}/{attr_key}/{attr_val}", a.byAttribute)
2020-03-31 08:37:10 +00:00
// attaching /-/(ready,healthy)
2020-11-09 13:43:23 +00:00
attachHealthy(r, a.pool.Status)
2020-03-31 08:37:10 +00:00
// enable metrics
if a.cfg.GetBool(cmdMetrics) {
2020-05-12 08:19:07 +00:00
a.log.Info("enabled /metrics/")
2020-03-31 08:37:10 +00:00
attachMetrics(r, a.wlog)
}
// enable pprof
if a.cfg.GetBool(cmdPprof) {
2020-05-12 08:19:07 +00:00
a.log.Info("enabled /debug/pprof/")
2020-03-31 08:37:10 +00:00
attachProfiler(r)
}
bind := a.cfg.GetString(cfgListenAddress)
2020-03-31 08:37:10 +00:00
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))
}
}