xk6-frostfs/internal/native/native.go

package native

import (
	"fmt"
	"math"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
	frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/stats"
	"github.com/google/uuid"
	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
	"go.k6.io/k6/js/modules"
	"go.k6.io/k6/metrics"
)

// RootModule is the global module object type. It is instantiated once per test
// run and will be used to create k6/x/frostfs/native module instances for each VU.
type RootModule struct{}

// Native represents an instance of the module for every VU.
type Native struct {
	vu modules.VU
}

// Ensure the interfaces are implemented correctly.
var (
	_ modules.Instance = &Native{}
	_ modules.Module   = &RootModule{}

	objPutSuccess, objPutFails, objPutDuration, objPutData *metrics.Metric
	objGetSuccess, objGetFails, objGetDuration, objGetData *metrics.Metric
	objDeleteSuccess, objDeleteFails, objDeleteDuration    *metrics.Metric
	cnrPutTotal, cnrPutFails, cnrPutDuration               *metrics.Metric
)

func init() {
	modules.Register("k6/x/frostfs/native", new(RootModule))
}

// NewModuleInstance implements the modules.Module interface and returns
// a new instance for each VU.
func (r *RootModule) NewModuleInstance(vu modules.VU) modules.Instance {
	mi := &Native{vu: vu}
	return mi
}

// Exports implements the modules.Instance interface and returns the exports
// of the JS module.
func (n *Native) Exports() modules.Exports {
	return modules.Exports{Default: n}
}

func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTimeout int, prepareLocally bool, maxObjSize int) (*Client, error) {
	var (
		cli client.Client
		pk  *keys.PrivateKey
		err error
	)

	if maxObjSize < 0 {
		return nil, fmt.Errorf("max object size value must be positive")
	}

	pk, err = keys.NewPrivateKey()
	if len(hexPrivateKey) != 0 {
		pk, err = keys.NewPrivateKeyFromHex(hexPrivateKey)
	}
	if err != nil {
		return nil, fmt.Errorf("invalid key: %w", err)
	}

	var prmInit client.PrmInit
	prmInit.Key = pk.PrivateKey
	cli.Init(prmInit)

	var prmDial client.PrmDial
	prmDial.Endpoint = endpoint

	if dialTimeout > 0 {
		prmDial.DialTimeout = time.Duration(dialTimeout) * time.Second
	}

	if streamTimeout > 0 {
		prmDial.StreamTimeout = time.Duration(streamTimeout) * time.Second
	}

	err = cli.Dial(n.vu.Context(), prmDial)
	if err != nil {
		return nil, fmt.Errorf("dial endpoint: %s %w", endpoint, err)
	}

	// generate session token
	exp := uint64(math.MaxUint64)
	sessionResp, err := cli.SessionCreate(n.vu.Context(), client.PrmSessionCreate{
		Expiration: exp,
	})
	if err != nil {
		return nil, fmt.Errorf("dial endpoint: %s %w", endpoint, err)
	}

	var id uuid.UUID
	err = id.UnmarshalBinary(sessionResp.ID())
	if err != nil {
		return nil, fmt.Errorf("session token: %w", err)
	}

	var key frostfsecdsa.PublicKey
	err = key.Decode(sessionResp.PublicKey())
	if err != nil {
		return nil, fmt.Errorf("invalid public session key: %w", err)
	}

	var tok session.Object

	tok.SetID(id)
	tok.SetAuthKey(&key)
	tok.SetExp(exp)

	if prepareLocally && maxObjSize > 0 {
		res, err := cli.NetworkInfo(n.vu.Context(), client.PrmNetworkInfo{})
		if err != nil {
			return nil, err
		}
		if uint64(maxObjSize) > res.Info().MaxObjectSize() {
			return nil, fmt.Errorf("max object size must be not greater than %d bytes", res.Info().MaxObjectSize())
		}
	}

	// register metrics

	objPutSuccess, _ = stats.Registry.NewMetric("frostfs_obj_put_success", metrics.Counter)
	objPutFails, _ = stats.Registry.NewMetric("frostfs_obj_put_fails", metrics.Counter)
	objPutDuration, _ = stats.Registry.NewMetric("frostfs_obj_put_duration", metrics.Trend, metrics.Time)
	objPutData, _ = stats.Registry.NewMetric("frostfs_obj_put_bytes", metrics.Counter, metrics.Data)

	objGetSuccess, _ = stats.Registry.NewMetric("frostfs_obj_get_success", metrics.Counter)
	objGetFails, _ = stats.Registry.NewMetric("frostfs_obj_get_fails", metrics.Counter)
	objGetDuration, _ = stats.Registry.NewMetric("frostfs_obj_get_duration", metrics.Trend, metrics.Time)
	objGetData, _ = stats.Registry.NewMetric("frostfs_obj_get_bytes", metrics.Counter, metrics.Data)

	objDeleteSuccess, _ = stats.Registry.NewMetric("frostfs_obj_delete_success", metrics.Counter)
	objDeleteFails, _ = stats.Registry.NewMetric("frostfs_obj_delete_fails", metrics.Counter)
	objDeleteDuration, _ = stats.Registry.NewMetric("frostfs_obj_delete_duration", metrics.Trend, metrics.Time)

	cnrPutTotal, _ = stats.Registry.NewMetric("frostfs_cnr_put_total", metrics.Counter)
	cnrPutFails, _ = stats.Registry.NewMetric("frostfs_cnr_put_fails", metrics.Counter)
	cnrPutDuration, _ = stats.Registry.NewMetric("frostfs_cnr_put_duration", metrics.Trend, metrics.Time)

	return &Client{
		vu:             n.vu,
		key:            pk.PrivateKey,
		tok:            tok,
		cli:            &cli,
		prepareLocally: prepareLocally,
		maxObjSize:     uint64(maxObjSize),
	}, nil
}