mkadilov/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#1161] node: Remove notification functionality

Browse source 
It is unused and will be reworked in future.

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
Evgenii Stratonikov 2024-06-04 15:31:47 +03:00 • committed by Evgenii Stratonikov
parent 67b3002743
commit a849236b68
17 changed files with 0 additions and 678 deletions
Download patch file Download diff file Expand all files Collapse all files

7
cmd/frostfs-node/config.go
View file

@ -474,7 +474,6 @@ type cfg struct {
cfgNetmap cfgNetmap
cfgControlService cfgControlService
cfgObject cfgObject
cfgNotifications cfgNotifications
}
// ReadCurrentNetMap reads network map which has been cached at the
@ -633,12 +632,6 @@ type cfgObject struct {
skipSessionTokenIssuerVerification bool
}
type cfgNotifications struct {
enabled bool
nw notificationWriter
defaultTopic string
}
type cfgLocalStorage struct {
localStorage *engine.StorageEngine
}

80
cmd/frostfs-node/config/node/config.go
View file

@ -5,7 +5,6 @@ import (
"io/fs"
"os"
"strconv"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network"
@ -25,12 +24,6 @@ type PersistentStateConfig struct {
cfg *config.Config
}
// NotificationConfig is a wrapper over "notification" config section
// which provides access to object notification configuration of node.
type NotificationConfig struct {
cfg *config.Config
}
// PersistentPolicyRulesConfig is a wrapper over "persistent_policy_rules" config section
// which provides access to persistent policy rules storage configuration of node.
type PersistentPolicyRulesConfig struct {
@ -41,16 +34,12 @@ const (
subsection = "node"
persistentSessionsSubsection = "persistent_sessions"
persistentStateSubsection = "persistent_state"
notificationSubsection = "notification"
persistentPolicyRulesSubsection = "persistent_policy_rules"
attributePrefix = "attribute"
// PersistentStatePathDefault is a default path for persistent state file.
PersistentStatePathDefault = ".frostfs-storage-state"
// NotificationTimeoutDefault is a default timeout for object notification operation.
NotificationTimeoutDefault = 5 * time.Second
)
// Key returns the value of "key" config parameter
@ -185,75 +174,6 @@ func (p PersistentStateConfig) Path() string {
return PersistentStatePathDefault
}
// Notification returns structure that provides access to "notification"
// subsection of "node" section.
func Notification(c *config.Config) NotificationConfig {
return NotificationConfig{
c.Sub(subsection).Sub(notificationSubsection),
}
}
// Enabled returns the value of "enabled" config parameter from "notification"
// subsection of "node" section.
//
// Returns false if the value is not presented.
func (n NotificationConfig) Enabled() bool {
return config.BoolSafe(n.cfg, "enabled")
}
// DefaultTopic returns the value of "default_topic" config parameter from
// "notification" subsection of "node" section.
//
// Returns empty string if the value is not presented.
func (n NotificationConfig) DefaultTopic() string {
return config.StringSafe(n.cfg, "default_topic")
}
// Endpoint returns the value of "endpoint" config parameter from "notification"
// subsection of "node" section.
//
// Returns empty string if the value is not presented.
func (n NotificationConfig) Endpoint() string {
return config.StringSafe(n.cfg, "endpoint")
}
// Timeout returns the value of "timeout" config parameter from "notification"
// subsection of "node" section.
//
// Returns NotificationTimeoutDefault if the value is not positive.
func (n NotificationConfig) Timeout() time.Duration {
v := config.DurationSafe(n.cfg, "timeout")
if v > 0 {
return v
}
return NotificationTimeoutDefault
}
// CertPath returns the value of "certificate_path" config parameter from "notification"
// subsection of "node" section.
//
// Returns empty string if the value is not presented.
func (n NotificationConfig) CertPath() string {
return config.StringSafe(n.cfg, "certificate")
}
// KeyPath returns the value of "key_path" config parameter from
// "notification" subsection of "node" section.
//
// Returns empty string if the value is not presented.
func (n NotificationConfig) KeyPath() string {
return config.StringSafe(n.cfg, "key")
}
// CAPath returns the value of "ca_path" config parameter from
// "notification" subsection of "node" section.
//
// Returns empty string if the value is not presented.
func (n NotificationConfig) CAPath() string {
return config.StringSafe(n.cfg, "ca")
}
const (
// PermDefault is a default permission bits for local override storage file.
PermDefault = 0o644

29
cmd/frostfs-node/config/node/config_test.go
View file

@ -2,7 +2,6 @@ package nodeconfig
import (
"testing"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config"
configtest "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/test"
@ -33,25 +32,11 @@ func TestNodeSection(t *testing.T) {
relay := Relay(empty)
persisessionsPath := PersistentSessions(empty).Path()
persistatePath := PersistentState(empty).Path()
notificationDefaultEnabled := Notification(empty).Enabled()
notificationDefaultEndpoint := Notification(empty).Endpoint()
notificationDefaultTimeout := Notification(empty).Timeout()
notificationDefaultTopic := Notification(empty).DefaultTopic()
notificationDefaultCertPath := Notification(empty).CertPath()
notificationDefaultKeyPath := Notification(empty).KeyPath()
notificationDefaultCAPath := Notification(empty).CAPath()
require.Empty(t, attribute)
require.Equal(t, false, relay)
require.Equal(t, "", persisessionsPath)
require.Equal(t, PersistentStatePathDefault, persistatePath)
require.Equal(t, false, notificationDefaultEnabled)
require.Equal(t, "", notificationDefaultEndpoint)
require.Equal(t, NotificationTimeoutDefault, notificationDefaultTimeout)
require.Equal(t, "", notificationDefaultTopic)
require.Equal(t, "", notificationDefaultCertPath)
require.Equal(t, "", notificationDefaultKeyPath)
require.Equal(t, "", notificationDefaultCAPath)
})
const path = "../../../../config/example/node"
@ -64,13 +49,6 @@ func TestNodeSection(t *testing.T) {
wKey := Wallet(c)
persisessionsPath := PersistentSessions(c).Path()
persistatePath := PersistentState(c).Path()
notificationEnabled := Notification(c).Enabled()
notificationEndpoint := Notification(c).Endpoint()
notificationTimeout := Notification(c).Timeout()
notificationDefaultTopic := Notification(c).DefaultTopic()
notificationCertPath := Notification(c).CertPath()
notificationKeyPath := Notification(c).KeyPath()
notificationCAPath := Notification(c).CAPath()
expectedAddr := []struct {
str string
@ -122,13 +100,6 @@ func TestNodeSection(t *testing.T) {
require.Equal(t, "/sessions", persisessionsPath)
require.Equal(t, "/state", persistatePath)
require.Equal(t, true, notificationEnabled)
require.Equal(t, "tls://localhost:4222", notificationEndpoint)
require.Equal(t, 6*time.Second, notificationTimeout)
require.Equal(t, "topic", notificationDefaultTopic)
require.Equal(t, "/cert/path", notificationCertPath)
require.Equal(t, "/key/path", notificationKeyPath)
require.Equal(t, "/ca/path", notificationCAPath)
}
configtest.ForEachFileType(path, fileConfigTest)

2
cmd/frostfs-node/main.go
View file

@ -111,7 +111,6 @@ func initApp(ctx context.Context, c *cfg) {
initAndLog(c, "accounting", func(c *cfg) { initAccountingService(ctx, c) })
initAndLog(c, "container", func(c *cfg) { initContainerService(ctx, c) })
initAndLog(c, "session", initSessionService)
initAndLog(c, "notification", func(c *cfg) { initNotifications(ctx, c) })
initAndLog(c, "object", initObjectService)
initAndLog(c, "tree", initTreeService)
initAndLog(c, "apemanager", initAPEManagerService)
@ -143,7 +142,6 @@ func stopAndLog(c *cfg, name string, stopper func() error) {
}
func bootUp(ctx context.Context, c *cfg) {
runAndLog(ctx, c, "NATS", true, connectNats)
runAndLog(ctx, c, "gRPC", false, func(_ context.Context, c *cfg) { serveGRPC(c) })
runAndLog(ctx, c, "notary", true, makeAndWaitNotaryDeposit)

172
cmd/frostfs-node/notificator.go
View file

@ -1,172 +0,0 @@
package main
import (
"context"
"encoding/hex"
"fmt"
nodeconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/node"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event/netmap"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/notificator"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/notificator/nats"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"go.uber.org/zap"
)
type notificationSource struct {
e *engine.StorageEngine
l *logger.Logger
defaultTopic string
}
func (n *notificationSource) Iterate(ctx context.Context, epoch uint64, handler func(topic string, addr oid.Address)) {
log := n.l.With(zap.Uint64("epoch", epoch))
listRes, err := n.e.ListContainers(ctx, engine.ListContainersPrm{})
if err != nil {
log.Error(logs.FrostFSNodeNotificatorCouldNotListContainers, zap.Error(err))
return
}
filters := objectSDK.NewSearchFilters()
filters.AddNotificationEpochFilter(epoch)
var selectPrm engine.SelectPrm
selectPrm.WithFilters(filters)
for _, c := range listRes.Containers() {
selectPrm.WithContainerID(c)
selectRes, err := n.e.Select(ctx, selectPrm)
if err != nil {
log.Error(logs.FrostFSNodeNotificatorCouldNotSelectObjectsFromContainer,
zap.Stringer("cid", c),
zap.Error(err),
)
continue
}
for _, a := range selectRes.AddressList() {
err = n.processAddress(ctx, a, handler)
if err != nil {
log.Error(logs.FrostFSNodeNotificatorCouldNotProcessObject,
zap.Stringer("address", a),
zap.Error(err),
)
continue
}
}
}
log.Debug(logs.FrostFSNodeNotificatorFinishedProcessingObjectNotifications)
}
func (n *notificationSource) processAddress(
ctx context.Context,
a oid.Address,
h func(topic string, addr oid.Address),
) error {
var prm engine.HeadPrm
prm.WithAddress(a)
res, err := n.e.Head(ctx, prm)
if err != nil {
return err
}
ni, err := res.Header().NotificationInfo()
if err != nil {
return fmt.Errorf("could not retrieve notification topic from object: %w", err)
}
topic := ni.Topic()
if topic == "" {
topic = n.defaultTopic
}
h(topic, a)
return nil
}
type notificationWriter struct {
l *logger.Logger
w *nats.Writer
}
func (n notificationWriter) Notify(topic string, address oid.Address) {
if err := n.w.Notify(topic, address); err != nil {
n.l.Warn(logs.FrostFSNodeCouldNotWriteObjectNotification,
zap.Stringer("address", address),
zap.String("topic", topic),
zap.Error(err),
)
}
}
func initNotifications(ctx context.Context, c *cfg) {
if nodeconfig.Notification(c.appCfg).Enabled() {
topic := nodeconfig.Notification(c.appCfg).DefaultTopic()
pubKey := hex.EncodeToString(c.cfgNodeInfo.localInfo.PublicKey())
if topic == "" {
topic = pubKey
}
natsSvc := nats.New(
nats.WithConnectionName("FrostFS Storage Node: "+pubKey), // connection name is used in the server side logs
nats.WithTimeout(nodeconfig.Notification(c.appCfg).Timeout()),
nats.WithClientCert(
nodeconfig.Notification(c.appCfg).CertPath(),
nodeconfig.Notification(c.appCfg).KeyPath(),
),
nats.WithRootCA(nodeconfig.Notification(c.appCfg).CAPath()),
nats.WithLogger(c.log),
)
c.cfgNotifications = cfgNotifications{
enabled: true,
nw: notificationWriter{
l: c.log,
w: natsSvc,
},
defaultTopic: topic,
}
n := notificator.New(new(notificator.Prm).
SetLogger(c.log).
SetNotificationSource(
&notificationSource{
e: c.cfgObject.cfgLocalStorage.localStorage,
l: c.log,
defaultTopic: topic,
}).
SetWriter(c.cfgNotifications.nw),
)
addNewEpochAsyncNotificationHandler(c, func(e event.Event) {
ev := e.(netmap.NewEpoch)
n.ProcessEpoch(ctx, ev.EpochNumber())
})
}
}
func connectNats(ctx context.Context, c *cfg) {
if !c.cfgNotifications.enabled {
return
}
endpoint := nodeconfig.Notification(c.appCfg).Endpoint()
err := c.cfgNotifications.nw.w.Connect(ctx, endpoint)
if err != nil {
panic(fmt.Sprintf("could not connect to a nats endpoint %s: %v", endpoint, err))
}
c.log.Info(logs.NatsConnectedToEndpoint, zap.String("endpoint", endpoint))
}

51
cmd/frostfs-node/object.go
View file

@ -14,7 +14,6 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
containercore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
morphClient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client"
cntClient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/container"
@ -333,15 +332,6 @@ func createPutSvc(c *cfg, keyStorage *util.KeyStorage, irFetcher *cachedIRFetche
engine: ls,
}
if c.cfgNotifications.enabled {
os = engineWithNotifications{
base: os,
nw: c.cfgNotifications.nw,
ns: c.cfgNetmap.state,
defaultTopic: c.cfgNotifications.defaultTopic,
}
}
return putsvc.NewService(
keyStorage,
c.putClientCache,
@ -503,47 +493,6 @@ func (s *morphEACLFetcher) GetEACL(cnr cid.ID) (*containercore.EACL, error) {
return eaclInfo, nil
}
type engineWithNotifications struct {
base putsvc.ObjectStorage
nw notificationWriter
ns netmap.State
defaultTopic string
}
func (e engineWithNotifications) IsLocked(ctx context.Context, address oid.Address) (bool, error) {
return e.base.IsLocked(ctx, address)
}
func (e engineWithNotifications) Delete(ctx context.Context, tombstone oid.Address, toDelete []oid.ID) error {
return e.base.Delete(ctx, tombstone, toDelete)
}
func (e engineWithNotifications) Lock(ctx context.Context, locker oid.Address, toLock []oid.ID) error {
return e.base.Lock(ctx, locker, toLock)
}
func (e engineWithNotifications) Put(ctx context.Context, o *objectSDK.Object) error {
if err := e.base.Put(ctx, o); err != nil {
return err
}
ni, err := o.NotificationInfo()
if err == nil {
if epoch := ni.Epoch(); epoch == 0 || epoch == e.ns.CurrentEpoch() {
topic := ni.Topic()
if topic == "" {
topic = e.defaultTopic
}
e.nw.Notify(topic, objectCore.AddressOf(o))
}
}
return nil
}
type engineWithoutNotifications struct {
engine *engine.StorageEngine
}