pkg/db/backend.go

/*
 * Copyright 2018-2024 Open Networking Foundation (ONF) and the ONF Contributors

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 * http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package db

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore"
	"github.com/opencord/voltha-lib-go/v7/pkg/log"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const (
	// Default Minimal Interval for posting alive state of backend kvstore on Liveness Channel
	DefaultLivenessChannelInterval = time.Second * 30
)

// Backend structure holds details for accessing the kv store
type Backend struct {
	Client                  kvstore.Client
	StoreType               string
	Timeout                 time.Duration
	Address                 string
	PathPrefix              string
	alive                   bool // Is this backend connection alive?
	livenessMutex           sync.Mutex
	liveness                chan bool     // channel to post alive state
	LivenessChannelInterval time.Duration // regularly push alive state beyond this interval
	lastLivenessTime        time.Time     // Instant of last alive state push
}

// NewBackend creates a new instance of a Backend structure
func NewBackend(ctx context.Context, storeType string, address string, timeout time.Duration, pathPrefix string) *Backend {
	var err error

	b := &Backend{
		StoreType:               storeType,
		Address:                 address,
		Timeout:                 timeout,
		LivenessChannelInterval: DefaultLivenessChannelInterval,
		PathPrefix:              pathPrefix,
		alive:                   false, // connection considered down at start
	}

	if b.Client, err = b.newClient(ctx, address, timeout); err != nil {
		logger.Errorw(ctx, "failed-to-create-kv-client",
			log.Fields{
				"type": storeType, "address": address,
				"timeout": timeout, "prefix": pathPrefix,
				"error": err.Error(),
			})
	}

	return b
}

func (b *Backend) newClient(ctx context.Context, address string, timeout time.Duration) (kvstore.Client, error) {
	switch b.StoreType {
	case "redis":
		return kvstore.NewRedisClient(address, timeout, false)
	case "redis-sentinel":
		return kvstore.NewRedisClient(address, timeout, true)
	case "etcd":
		return kvstore.NewEtcdClient(ctx, address, timeout, log.WarnLevel)
	}
	return nil, errors.New("unsupported-kv-store")
}

func (b *Backend) makePath(ctx context.Context, key string) string {
	path := fmt.Sprintf("%s/%s", b.PathPrefix, key)
	logger.Debugw(ctx, "make-path", log.Fields{"key": key, "path": path})
	return path
}

func (b *Backend) updateLiveness(ctx context.Context, alive bool) {
	// Periodically push stream of liveness data to the channel,
	// so that in a live state, the core does not timeout and
	// send a forced liveness message. Push alive state if the
	// last push to channel was beyond livenessChannelInterval
	b.livenessMutex.Lock()
	defer b.livenessMutex.Unlock()
	if b.liveness != nil {
		if b.alive != alive {
			logger.Debug(ctx, "update-liveness-channel-reason-change")
			b.liveness <- alive
			b.lastLivenessTime = time.Now()
		} else if time.Since(b.lastLivenessTime) > b.LivenessChannelInterval {
			logger.Debug(ctx, "update-liveness-channel-reason-interval")
			b.liveness <- alive
			b.lastLivenessTime = time.Now()
		}
	}

	// Emit log message only for alive state change
	if b.alive != alive {
		logger.Debugw(ctx, "change-kvstore-alive-status", log.Fields{"alive": alive})
		b.alive = alive
	}
}

// Perform a dummy Key Lookup on kvstore to test Connection Liveness and
// post on Liveness channel
func (b *Backend) PerformLivenessCheck(ctx context.Context) bool {
	alive := b.Client.IsConnectionUp(ctx)
	logger.Debugw(ctx, "kvstore-liveness-check-result", log.Fields{"alive": alive})

	b.updateLiveness(ctx, alive)
	return alive
}

// Enable the liveness monitor channel. This channel will report
// a "true" or "false" on every kvstore operation which indicates whether
// or not the connection is still Live. This channel is then picked up
// by the service (i.e. rw_core / ro_core) to update readiness status
// and/or take other actions.
func (b *Backend) EnableLivenessChannel(ctx context.Context) chan bool {
	logger.Debug(ctx, "enable-kvstore-liveness-channel")
	b.livenessMutex.Lock()
	defer b.livenessMutex.Unlock()
	if b.liveness == nil {
		b.liveness = make(chan bool, 10)
		b.liveness <- b.alive
		b.lastLivenessTime = time.Now()
	}

	return b.liveness
}

// Extract Alive status of Kvstore based on type of error
func (b *Backend) isErrorIndicatingAliveKvstore(err error) bool {
	// Alive unless observed an error indicating so
	alive := true

	if err != nil {

		// timeout indicates kvstore not reachable/alive
		if err == context.DeadlineExceeded {
			alive = false
		}

		// Need to analyze client-specific errors based on backend type
		if b.StoreType == "etcd" {

			// For etcd backend, consider not-alive only for errors indicating
			// timedout request or unavailable/corrupted cluster. For all remaining
			// error codes listed in https://godoc.org/google.golang.org/grpc/codes#Code,
			// we would not infer a not-alive backend because such a error may also
			// occur due to bad client requests or sequence of operations
			switch status.Code(err) {
			case codes.DeadlineExceeded:
				fallthrough
			case codes.Unavailable:
				fallthrough
			case codes.DataLoss:
				alive = false
			}
		}
	}

	return alive
}

// List retrieves one or more items that match the specified key
func (b *Backend) List(ctx context.Context, key string) (map[string]*kvstore.KVPair, error) {
	span, ctx := log.CreateChildSpan(ctx, "kvs-list")
	defer span.Finish()

	formattedPath := b.makePath(ctx, key)
	logger.Debugw(ctx, "listing-key", log.Fields{"key": key, "path": formattedPath})

	pair, err := b.Client.List(ctx, formattedPath)

	b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(err))

	return pair, err
}

// Get retrieves an item that matches the specified key
func (b *Backend) Get(ctx context.Context, key string) (*kvstore.KVPair, error) {
	span, ctx := log.CreateChildSpan(ctx, "kvs-get")
	defer span.Finish()

	formattedPath := b.makePath(ctx, key)
	logger.Debugw(ctx, "getting-key", log.Fields{"key": key, "path": formattedPath})

	pair, err := b.Client.Get(ctx, formattedPath)

	b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(err))

	return pair, err
}

// GetWithPrefix retrieves one or more items that match the specified key prefix
func (b *Backend) GetWithPrefix(ctx context.Context, prefixKey string) (map[string]*kvstore.KVPair, error) {
	span, ctx := log.CreateChildSpan(ctx, "kvs-get-with-prefix")
	defer span.Finish()

	formattedPath := b.makePath(ctx, prefixKey)
	logger.Debugw(ctx, "get-entries-matching-prefix-key", log.Fields{"key": prefixKey, "path": formattedPath})

	pair, err := b.Client.GetWithPrefix(ctx, formattedPath)

	b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(err))

	return pair, err
}

// GetWithPrefixKeysOnly retrieves one or more keys that match the specified key prefix
func (b *Backend) GetWithPrefixKeysOnly(ctx context.Context, prefixKey string) ([]string, error) {
	span, ctx := log.CreateChildSpan(ctx, "kvs-get-with-prefix")
	defer span.Finish()

	formattedPath := b.makePath(ctx, prefixKey)
	logger.Debugw(ctx, "get-keys-entries-matching-prefix-key", log.Fields{"key": prefixKey, "path": formattedPath})

	pair, err := b.Client.GetWithPrefixKeysOnly(ctx, formattedPath)

	b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(err))

	return pair, err
}

// Put stores an item value under the specifed key
func (b *Backend) Put(ctx context.Context, key string, value interface{}) error {
	span, ctx := log.CreateChildSpan(ctx, "kvs-put")
	defer span.Finish()

	formattedPath := b.makePath(ctx, key)
	logger.Debugw(ctx, "putting-key", log.Fields{"key": key, "path": formattedPath})

	err := b.Client.Put(ctx, formattedPath, value)

	b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(err))

	return err
}

// Delete removes an item under the specified key
func (b *Backend) Delete(ctx context.Context, key string) error {
	span, ctx := log.CreateChildSpan(ctx, "kvs-delete")
	defer span.Finish()

	formattedPath := b.makePath(ctx, key)
	logger.Debugw(ctx, "deleting-key", log.Fields{"key": key, "path": formattedPath})

	err := b.Client.Delete(ctx, formattedPath)

	b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(err))

	return err
}

func (b *Backend) DeleteWithPrefix(ctx context.Context, prefixKey string) error {
	span, ctx := log.CreateChildSpan(ctx, "kvs-delete-with-prefix")
	defer span.Finish()

	formattedPath := b.makePath(ctx, prefixKey)
	logger.Debugw(ctx, "deleting-prefix-key", log.Fields{"key": prefixKey, "path": formattedPath})

	err := b.Client.DeleteWithPrefix(ctx, formattedPath)

	b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(err))

	return err
}

// CreateWatch starts watching events for the specified key
func (b *Backend) CreateWatch(ctx context.Context, key string, withPrefix bool) chan *kvstore.Event {
	span, ctx := log.CreateChildSpan(ctx, "kvs-create-watch")
	defer span.Finish()

	formattedPath := b.makePath(ctx, key)
	logger.Debugw(ctx, "creating-key-watch", log.Fields{"key": key, "path": formattedPath})

	return b.Client.Watch(ctx, formattedPath, withPrefix)
}

// DeleteWatch stops watching events for the specified key
func (b *Backend) DeleteWatch(ctx context.Context, key string, ch chan *kvstore.Event) {
	span, ctx := log.CreateChildSpan(ctx, "kvs-delete-watch")
	defer span.Finish()

	formattedPath := b.makePath(ctx, key)
	logger.Debugw(ctx, "deleting-key-watch", log.Fields{"key": key, "path": formattedPath})

	b.Client.CloseWatch(ctx, formattedPath, ch)
}