frostfs-http-gw/main.go

package main

import (
	"context"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/labstack/echo/v4"
	"github.com/nspcc-dev/neofs-proto/object"
	"github.com/nspcc-dev/neofs-proto/refs"
	"github.com/nspcc-dev/neofs-proto/service"
	"github.com/pkg/errors"
	"github.com/spf13/viper"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/keepalive"
)

type config struct {
	log     *zap.Logger
	timeout time.Duration
	cli     object.ServiceClient
}

var defaultConfig = strings.NewReader(`
request_timeout: 5s
connect_timeout: 30s
listen_address: :8082
neofs_node_addr: :8080

logger:
  level: debug
  format: console
  trace_level: fatal
  no_disclaimer: true
  sampling:
    initial: 1000
    thereafter: 1000

keepalive:
  time: 100ms
  timeout: 10s
  permit_without_stream: true
`)

func main() {
	v := viper.New()
	v.AutomaticEnv()
	v.SetEnvPrefix("GW")
	v.SetConfigType("yaml")
	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))

	// set defaults:
	v.Set("app.name", "neofs-gw")
	v.Set("app.version", Version)

	if err := v.ReadConfig(defaultConfig); err != nil {
		panic(err)
	}

	log, err := newLogger(v)
	if err != nil {
		panic(err)
	}

	log.Info("running application", zap.String("version", v.GetString("app.version")))

	var (
		cfg   = new(config)
		grace = newGracefulContext(log)
	)

	cfg.log = log
	cfg.timeout = v.GetDuration("request_timeout")
	ctx, cancel := context.WithTimeout(grace, v.GetDuration("connect_timeout"))
	defer cancel()

	conn, err := grpc.DialContext(ctx, v.GetString("neofs_node_addr"),
		grpc.WithInsecure(),
		grpc.WithKeepaliveParams(keepalive.ClientParameters{
			Time:                v.GetDuration("keepalive.time"),
			Timeout:             v.GetDuration("keepalive.timeout"),
			PermitWithoutStream: v.GetBool("keepalive.permit_without_stream"),
		}),
	)
	if err != nil {
		log.Panic("could not connect to neofs-node",
			zap.String("neofs-node", v.GetString("neofs_node_addr")),
			zap.Error(err))
	}

	ctx, cancel = context.WithCancel(grace)
	defer cancel()

	go checkConnection(ctx, conn, log)
	cfg.cli = object.NewServiceClient(conn)

	e := echo.New()
	e.Debug = false
	e.HidePort = true
	e.HideBanner = true

	e.GET("/:cid/:oid", cfg.receiveFile)
	go func() {
		log.Info("run gateway server",
			zap.String("address", v.GetString("listen_address")))
		if err := e.Start(v.GetString("listen_address")); err != nil {
			log.Panic("could not start server", zap.Error(err))
		}
	}()

	<-ctx.Done()

	ctx, cancel = context.WithTimeout(context.TODO(), time.Second*30)
	defer cancel()

	if err := e.Shutdown(ctx); err != nil {
		log.Panic("could not stop server", zap.Error(err))
	}
}

func checkConnection(ctx context.Context, conn *grpc.ClientConn, log *zap.Logger) {
	tick := time.NewTicker(time.Second)
loop:
	for {
		select {
		case <-ctx.Done():
			break loop
		case <-tick.C:
			switch state := conn.GetState(); state {
			case connectivity.Idle, connectivity.Connecting, connectivity.Ready:
				// It's ok..
			default:
				log.Panic("could not establish connection",
					zap.Stringer("state", state),
					zap.Any("connection", conn.Target()))
			}
		}
	}

	tick.Stop()
}

func (cfg *config) receiveFile(c echo.Context) error {
	var (
		cid      refs.CID
		oid      refs.ObjectID
		obj      *object.Object
		download = c.QueryParam("download") != ""
	)

	cfg.log.Debug("try to fetch object from network",
		zap.String("cid", c.Param("cid")),
		zap.String("oid", c.Param("oid")))

	if err := cid.Parse(c.Param("cid")); err != nil {
		cfg.log.Error("wrong container id",
			zap.String("cid", c.Param("cid")),
			zap.String("oid", c.Param("oid")),
			zap.Error(err))

		return echo.NewHTTPError(
			http.StatusBadRequest,
			errors.Wrap(err, "wrong container id").Error(),
		)
	} else if err := oid.Parse(c.Param("oid")); err != nil {
		cfg.log.Error("wrong object id",
			zap.Stringer("cid", cid),
			zap.String("oid", c.Param("oid")),
			zap.Error(err))

		return echo.NewHTTPError(
			http.StatusBadRequest,
			errors.Wrap(err, "wrong object id").Error(),
		)
	}

	ctx, cancel := context.WithTimeout(context.Background(), cfg.timeout)
	defer cancel()

	cli, err := cfg.cli.Get(ctx, &object.GetRequest{
		Address: refs.Address{ObjectID: oid, CID: cid},
		TTL:     service.SingleForwardingTTL,
	})

	if err != nil {
		cfg.log.Error("could not prepare connection",
			zap.Stringer("cid", cid),
			zap.Stringer("oid", oid),
			zap.Error(err))

		return echo.NewHTTPError(
			// TODO: nginx doesn't return 500 errors from backend
			// http.StatusInternalServerError,
			http.StatusBadRequest,
			errors.Wrap(err, "could not prepare connection").Error(),
		)
	} else if obj, err = receiveObject(cli); err != nil {
		cfg.log.Error("could not receive object",
			zap.Stringer("cid", cid),
			zap.Stringer("oid", oid),
			zap.Error(err))

		if strings.Contains(err.Error(), object.ErrNotFound.Error()) {
			return echo.NewHTTPError(http.StatusNotFound, err.Error())
		}

		return echo.NewHTTPError(
			// TODO: nginx doesn't return 500 errors from backend
			// http.StatusInternalServerError,
			http.StatusBadRequest,
			errors.Wrap(err, "could not receive object").Error(),
		)
	}

	cfg.log.Info("object fetched successfully",
		zap.Stringer("cid", cid),
		zap.Stringer("oid", oid))

	c.Response().Header().Set("Content-Length", strconv.FormatUint(obj.SystemHeader.PayloadLength, 10))
	c.Response().Header().Set("x-object-id", obj.SystemHeader.ID.String())
	c.Response().Header().Set("x-owner-id", obj.SystemHeader.OwnerID.String())
	c.Response().Header().Set("x-container-id", obj.SystemHeader.CID.String())

	for i := range obj.Headers {
		if hdr := obj.Headers[i].GetUserHeader(); hdr != nil {
			c.Response().Header().Set("x-"+hdr.Key, hdr.Value)

			if hdr.Key == DropInFilenameHeader && download {
				c.Response().Header().Set("Content-Disposition", "attachment; filename="+hdr.Value)
			}
		}
	}

	return c.Blob(http.StatusOK,
		http.DetectContentType(obj.Payload),
		obj.Payload)
}

func receiveObject(cli object.Service_GetClient) (*object.Object, error) {
	var obj *object.Object
	for {
		resp, err := cli.Recv()
		if err != nil {
			if err == io.EOF {
				break
			}
			return nil, err
		} else if obj == nil {
			obj = resp.GetObject()
		}

		obj.Payload = append(obj.Payload, resp.GetChunk()...)
	}
	return obj, nil
}