path: root/src/k8splugin/internal/app/subscription.go

/*
 * Copyright © 2022 Orange
 *
 * 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 app

import (
	"bytes"
	"encoding/json"
	"io/ioutil"
	"net"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/onap/multicloud-k8s/src/k8splugin/internal/db"
	log "github.com/onap/multicloud-k8s/src/k8splugin/internal/logutils"
	"github.com/onap/multicloud-k8s/src/k8splugin/internal/rb"
	pkgerrors "github.com/pkg/errors"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/tools/cache"
)

// QueryStatus is what is returned when status is queried for an instance
type StatusSubscription struct {
	Name              string                 `json:"name"`
	MinNotifyInterval int32                  `json:"min-notify-interval"`
	LastUpdateTime    time.Time              `json:"last-update-time"`
	CallbackUrl       string                 `json:"callback-url"`
	LastNotifyTime    time.Time              `json:"last-notify-time"`
	LastNotifyStatus  int32                  `json:"last-notify-status"`
	NotifyMetadata    map[string]interface{} `json:"metadata"`
}

type SubscriptionRequest struct {
	Name              string                 `json:"name"`
	MinNotifyInterval int32                  `json:"min-notify-interval"`
	NotifyMetadata    map[string]interface{} `json:"metadata"`
	CallbackUrl       string                 `json:"callback-url"`
}

// StatusSubscriptionKey is used as the primary key in the db
type StatusSubscriptionKey struct {
	InstanceId       string `json:"instanceid"`
	SubscriptionName string `json:"subscriptionname"`
}

// We will use json marshalling to convert to string to
// preserve the underlying structure.
func (dk StatusSubscriptionKey) String() string {
	out, err := json.Marshal(dk)
	if err != nil {
		return ""
	}

	return string(out)
}

// InstanceStatusSubClient implements InstanceStatusSubManager
type InstanceStatusSubClient struct {
	storeName string
	tagInst   string
}

func NewInstanceStatusSubClient() *InstanceStatusSubClient {
	return &InstanceStatusSubClient{
		storeName: "rbdef",
		tagInst:   "instanceStatusSub",
	}
}

type notifyResult struct {
	result int32
	time   time.Time
}

type resourceStatusDelta struct {
	Created  []ResourceStatus `json:"created"`
	Deleted  []ResourceStatus `json:"deleted"`
	Modified []ResourceStatus `json:"modified"`
}

type notifyRequestPayload struct {
	InstanceId   string                 `json:"instance-id"`
	Subscription string                 `json:"subscription-name"`
	Metadata     map[string]interface{} `json:"metadata"`
	Delta        resourceStatusDelta    `json:"resource-changes"`
}

func (rsd resourceStatusDelta) Delta() bool {
	return len(rsd.Created) > 0 || len(rsd.Deleted) > 0 || len(rsd.Modified) > 0
}

type notifyChannelData struct {
	instanceId   string
	subscription StatusSubscription
	action       string
	delta        resourceStatusDelta
	notifyResult chan notifyResult
}

type subscriptionWatch struct {
	watcherStop    chan struct{}
	lastUpdateTime time.Time
}

type subscriptionWatchManager struct {
	watchersStatus map[string]subscriptionWatch
}

type subscriptionNotifyManager struct {
	notifyLockMap  map[string]*sync.Mutex
	notifyChannel  map[string]chan notifyChannelData
	watchersStatus map[string]subscriptionWatchManager
	sync.Mutex
}

var subscriptionNotifyData = subscriptionNotifyManager{
	notifyLockMap:  map[string]*sync.Mutex{},
	notifyChannel:  map[string]chan notifyChannelData{},
	watchersStatus: map[string]subscriptionWatchManager{},
}

// InstanceStatusSubManager is an interface exposes the status subscription functionality
type InstanceStatusSubManager interface {
	Create(instanceId string, subDetails SubscriptionRequest) (StatusSubscription, error)
	Get(instanceId, subId string) (StatusSubscription, error)
	Update(instanceId, subId string, subDetails SubscriptionRequest) (StatusSubscription, error)
	List(instanceId string) ([]StatusSubscription, error)
	Delete(instanceId, subId string) error
	Cleanup(instanceId string) error
	RestoreWatchers()
}

// Create Status Subscription
func (iss *InstanceStatusSubClient) Create(instanceId string, subDetails SubscriptionRequest) (StatusSubscription, error) {

	_, err := iss.Get(instanceId, subDetails.Name)
	if err == nil {
		return StatusSubscription{}, pkgerrors.New("Subscription already exists")
	}

	lock, _, _ := getSubscriptionData(instanceId)

	key := StatusSubscriptionKey{
		InstanceId:       instanceId,
		SubscriptionName: subDetails.Name,
	}

	sub := StatusSubscription{
		Name:              subDetails.Name,
		MinNotifyInterval: subDetails.MinNotifyInterval,
		LastNotifyStatus:  0,
		CallbackUrl:       subDetails.CallbackUrl,
		LastUpdateTime:    time.Now(),
		LastNotifyTime:    time.Now(),
		NotifyMetadata:    subDetails.NotifyMetadata,
	}
	if sub.NotifyMetadata == nil {
		sub.NotifyMetadata = make(map[string]interface{})
	}

	err = iss.refreshWatchers(instanceId, subDetails.Name)
	if err != nil {
		return sub, pkgerrors.Wrap(err, "Creating Status Subscription DB Entry")
	}

	lock.Lock()
	defer lock.Unlock()

	err = db.DBconn.Create(iss.storeName, key, iss.tagInst, sub)
	if err != nil {
		return sub, pkgerrors.Wrap(err, "Creating Status Subscription DB Entry")
	}
	log.Info("Successfully created Status Subscription", log.Fields{
		"InstanceId":       instanceId,
		"SubscriptionName": subDetails.Name,
	})

	go runNotifyThread(instanceId, sub.Name)

	return sub, nil
}

// Get Status subscription
func (iss *InstanceStatusSubClient) Get(instanceId, subId string) (StatusSubscription, error) {
	lock, _, _ := getSubscriptionData(instanceId)
	// Acquire Mutex
	lock.Lock()
	defer lock.Unlock()
	key := StatusSubscriptionKey{
		InstanceId:       instanceId,
		SubscriptionName: subId,
	}
	DBResp, err := db.DBconn.Read(iss.storeName, key, iss.tagInst)
	if err != nil || DBResp == nil {
		return StatusSubscription{}, pkgerrors.Wrap(err, "Error retrieving Subscription data")
	}

	sub := StatusSubscription{}
	err = db.DBconn.Unmarshal(DBResp, &sub)
	if err != nil {
		return StatusSubscription{}, pkgerrors.Wrap(err, "Demarshalling Subscription Value")
	}
	return sub, nil
}

// Update status subscription
func (iss *InstanceStatusSubClient) Update(instanceId, subId string, subDetails SubscriptionRequest) (StatusSubscription, error) {
	sub, err := iss.Get(instanceId, subDetails.Name)
	if err != nil {
		return StatusSubscription{}, pkgerrors.Wrap(err, "Subscription does not exist")
	}

	lock, _, _ := getSubscriptionData(instanceId)

	key := StatusSubscriptionKey{
		InstanceId:       instanceId,
		SubscriptionName: subDetails.Name,
	}

	sub.MinNotifyInterval = subDetails.MinNotifyInterval
	sub.CallbackUrl = subDetails.CallbackUrl
	sub.NotifyMetadata = subDetails.NotifyMetadata
	if sub.NotifyMetadata == nil {
		sub.NotifyMetadata = make(map[string]interface{})
	}

	err = iss.refreshWatchers(instanceId, subDetails.Name)
	if err != nil {
		return sub, pkgerrors.Wrap(err, "Updating Status Subscription DB Entry")
	}

	lock.Lock()
	defer lock.Unlock()

	err = db.DBconn.Update(iss.storeName, key, iss.tagInst, sub)
	if err != nil {
		return sub, pkgerrors.Wrap(err, "Updating Status Subscription DB Entry")
	}
	log.Info("Successfully updated Status Subscription", log.Fields{
		"InstanceId":       instanceId,
		"SubscriptionName": subDetails.Name,
	})

	return sub, nil
}

// Get list of status subscriptions
func (iss *InstanceStatusSubClient) List(instanceId string) ([]StatusSubscription, error) {

	lock, _, _ := getSubscriptionData(instanceId)
	// Acquire Mutex
	lock.Lock()
	defer lock.Unlock()
	// Retrieve info about created status subscriptions
	dbResp, err := db.DBconn.ReadAll(iss.storeName, iss.tagInst)
	if err != nil {
		if !strings.Contains(err.Error(), "Did not find any objects with tag") {
			return []StatusSubscription{}, pkgerrors.Wrap(err, "Getting Status Subscription data")
		}
	}
	subList := make([]StatusSubscription, 0)
	for key, value := range dbResp {
		if key != "" {
			subKey := StatusSubscriptionKey{}
			err = json.Unmarshal([]byte(key), &subKey)
			if err != nil {
				log.Error("Error demarshaling Status Subscription Key DB data", log.Fields{
					"error": err.Error(),
					"key":   key})
				return []StatusSubscription{}, pkgerrors.Wrap(err, "Demarshalling subscription key")
			}
			if subKey.InstanceId != instanceId {
				continue
			}
		}
		//value is a byte array
		if value != nil {
			sub := StatusSubscription{}
			err = db.DBconn.Unmarshal(value, &sub)
			if err != nil {
				log.Error("Error demarshaling Status Subscription DB data", log.Fields{
					"error": err.Error(),
					"key":   key})
			}
			subList = append(subList, sub)
		}
	}

	return subList, nil
}

// Delete status subscription
func (iss *InstanceStatusSubClient) Delete(instanceId, subId string) error {
	_, err := iss.Get(instanceId, subId)
	if err != nil {
		return pkgerrors.Wrap(err, "Subscription does not exist")
	}
	lock, _, watchers := getSubscriptionData(instanceId)
	// Acquire Mutex
	lock.Lock()
	defer lock.Unlock()

	close(watchers.watchersStatus[subId].watcherStop)
	delete(watchers.watchersStatus, subId)

	key := StatusSubscriptionKey{
		InstanceId:       instanceId,
		SubscriptionName: subId,
	}
	err = db.DBconn.Delete(iss.storeName, key, iss.tagInst)
	if err != nil {
		return pkgerrors.Wrap(err, "Removing Status Subscription in DB")
	}
	return nil
}

// Cleanup status subscriptions for instance
func (iss *InstanceStatusSubClient) Cleanup(instanceId string) error {
	subList, err := iss.List(instanceId)
	if err != nil {
		return err
	}

	for _, sub := range subList {
		err = iss.Delete(instanceId, sub.Name)
		if err != nil {
			log.Error("Error deleting ", log.Fields{
				"error": err.Error(),
				"key":   sub.Name})
		}
	}
	removeSubscriptionData(instanceId)
	return err
}

// Restore status subscriptions notify threads
func (iss *InstanceStatusSubClient) RestoreWatchers() {
	go func() {
		time.Sleep(time.Second * 10)
		log.Info("Restoring status subscription notifications", log.Fields{})
		v := NewInstanceClient()
		instances, err := v.List("", "", "")
		if err != nil {
			log.Error("Error reading instance list", log.Fields{
				"error": err.Error(),
			})
		}
		for _, instance := range instances {
			subList, err := iss.List(instance.ID)
			if err != nil {
				log.Error("Error reading subscription list for instance", log.Fields{
					"error":    err.Error(),
					"instance": instance.ID,
				})
				continue
			}

			for _, sub := range subList {
				err = iss.refreshWatchers(instance.ID, sub.Name)
				if err != nil {
					log.Error("Error on refreshing watchers", log.Fields{
						"error":        err.Error(),
						"instance":     instance.ID,
						"subscription": sub.Name,
					})
					continue
				}
				go runNotifyThread(instance.ID, sub.Name)
			}
		}
	}()
}

func (iss *InstanceStatusSubClient) refreshWatchers(instanceId, subId string) error {
	log.Info("REFRESH WATCHERS", log.Fields{
		"instance":     instanceId,
		"subscription": subId,
	})
	v := NewInstanceClient()
	k8sClient := KubernetesClient{}
	instance, err := v.Get(instanceId)
	if err != nil {
		return pkgerrors.Wrap(err, "Cannot get instance for notify thread")
	}
	profile, err := rb.NewProfileClient().Get(instance.Request.RBName, instance.Request.RBVersion,
		instance.Request.ProfileName)
	if err != nil {
		return pkgerrors.Wrap(err, "Unable to find Profile instance status")
	}
	err = k8sClient.Init(instance.Request.CloudRegion, instanceId)
	if err != nil {
		return pkgerrors.Wrap(err, "Cannot set k8s client for instance")
	}

	lock, _, watchers := getSubscriptionData(instanceId)
	// Acquire Mutex
	lock.Lock()
	defer lock.Unlock()
	watcher, ok := watchers.watchersStatus[subId]
	if ok {
		close(watcher.watcherStop)
	} else {
		watchers.watchersStatus[subId] = subscriptionWatch{
			lastUpdateTime: time.Now(),
		}
	}

	watcher.watcherStop = make(chan struct{})

	for _, gvk := range gvkListForInstance(instance, profile) {
		informer, _ := k8sClient.GetInformer(gvk)
		handlers := cache.ResourceEventHandlerFuncs{
			AddFunc: func(obj interface{}) {
				lock.Lock()
				watcher.lastUpdateTime = time.Now()
				watchers.watchersStatus[subId] = watcher
				lock.Unlock()
			},
			UpdateFunc: func(oldObj, obj interface{}) {
				lock.Lock()
				watcher.lastUpdateTime = time.Now()
				watchers.watchersStatus[subId] = watcher
				lock.Unlock()
			},
			DeleteFunc: func(obj interface{}) {
				lock.Lock()
				watcher.lastUpdateTime = time.Now()
				watchers.watchersStatus[subId] = watcher
				lock.Unlock()
			},
		}
		informer.AddEventHandler(handlers)
		go func(informer cache.SharedInformer, stopper chan struct{}, fields log.Fields) {
			log.Info("[START] Watcher", fields)
			informer.Run(stopper)
			log.Info("[STOP] Watcher", fields)
		}(informer, watcher.watcherStop, log.Fields{
			"Kind":         gvk.Kind,
			"Instance":     instanceId,
			"Subscription": subId,
		})
	}
	return nil
}

// Get the Mutex for the Subscription
func getSubscriptionData(instanceId string) (*sync.Mutex, chan notifyChannelData, subscriptionWatchManager) {
	var key string = instanceId
	subscriptionNotifyData.Lock()
	defer subscriptionNotifyData.Unlock()
	_, ok := subscriptionNotifyData.notifyLockMap[key]
	if !ok {
		subscriptionNotifyData.notifyLockMap[key] = &sync.Mutex{}
	}
	_, ok = subscriptionNotifyData.notifyChannel[key]
	if !ok {
		subscriptionNotifyData.notifyChannel[key] = make(chan notifyChannelData)
		go scheduleNotifications(instanceId, subscriptionNotifyData.notifyChannel[key])
		time.Sleep(time.Second * 5)
	}
	_, ok = subscriptionNotifyData.watchersStatus[key]
	if !ok {
		subscriptionNotifyData.watchersStatus[key] = subscriptionWatchManager{
			watchersStatus: make(map[string]subscriptionWatch),
		}
	}
	return subscriptionNotifyData.notifyLockMap[key], subscriptionNotifyData.notifyChannel[key], subscriptionNotifyData.watchersStatus[key]
}

func removeSubscriptionData(instanceId string) {
	var key string = instanceId
	subscriptionNotifyData.Lock()
	defer subscriptionNotifyData.Unlock()
	_, ok := subscriptionNotifyData.notifyLockMap[key]
	if ok {
		delete(subscriptionNotifyData.notifyLockMap, key)
	}
	_, ok = subscriptionNotifyData.notifyChannel[key]
	if ok {
		crl := notifyChannelData{
			instanceId: instanceId,
			action:     "STOP",
		}
		subscriptionNotifyData.notifyChannel[key] <- crl
		delete(subscriptionNotifyData.notifyChannel, key)
	}
	_, ok = subscriptionNotifyData.watchersStatus[key]
	if !ok {
		delete(subscriptionNotifyData.watchersStatus, key)
	}
}

// notify request timeout
func notifyTimeout(network, addr string) (net.Conn, error) {
	return net.DialTimeout(network, addr, time.Duration(time.Second*5))
}

// Per Subscription Go routine to send notification about status change
func scheduleNotifications(instanceId string, c chan notifyChannelData) {
	// Keep thread running
	log.Info("[START] status notify thread for ", log.Fields{
		"instance": instanceId,
	})
	for {
		data := <-c
		breakThread := false
		switch {
		case data.action == "NOTIFY":
			var result = notifyResult{}
			var err error = nil
			var notifyPayload = notifyRequestPayload{
				Delta:        data.delta,
				InstanceId:   data.instanceId,
				Subscription: data.subscription.Name,
				Metadata:     data.subscription.NotifyMetadata,
			}
			notifyBody, err := json.Marshal(notifyPayload)
			if err == nil {
				notifyBodyBuffer := bytes.NewBuffer(notifyBody)
				transport := http.Transport{
					Dial: notifyTimeout,
				}
				client := http.Client{
					Transport: &transport,
				}
				resp, errReq := client.Post(data.subscription.CallbackUrl, "application/json", notifyBodyBuffer)
				if errReq == nil {
					result.result = int32(resp.StatusCode)
					if resp.StatusCode >= 400 {
						respBody, _ := ioutil.ReadAll(resp.Body)
						log.Error("Status notification request failed", log.Fields{
							"instance": instanceId,
							"name":     data.subscription.Name,
							"url":      data.subscription.CallbackUrl,
							"code":     resp.StatusCode,
							"status":   resp.Status,
							"body":     string(respBody),
						})
						resp.Body.Close()
					}
				} else {
					err = errReq
				}
			}

			if err != nil {
				log.Error("Error for status notify thread", log.Fields{
					"instance": instanceId,
					"name":     data.subscription.Name,
					"err":      err.Error(),
				})
				result.result = 500
			}
			result.time = time.Now()

			data.notifyResult <- result

		case data.action == "STOP":
			breakThread = true
		}
		if breakThread {
			break
		}
	}
	log.Info("[STOP] status notify thread for ", log.Fields{
		"instance": instanceId,
	})
}

func gvkListForInstance(instance InstanceResponse, profile rb.Profile) []schema.GroupVersionKind {
	list := make([]schema.GroupVersionKind, 0)
	gvkMap := make(map[string]schema.GroupVersionKind)
	gvk := schema.FromAPIVersionAndKind("v1", "Pod")
	gvkMap[gvk.String()] = gvk
	for _, res := range instance.Resources {
		gvk = res.GVK
		_, ok := gvkMap[gvk.String()]
		if !ok {
			gvkMap[gvk.String()] = gvk
		}
	}
	for _, gvk := range profile.ExtraResourceTypes {
		_, ok := gvkMap[gvk.String()]
		if !ok {
			gvkMap[gvk.String()] = gvk
		}
	}
	for _, gvk := range gvkMap {
		list = append(list, gvk)
	}
	return list
}

func runNotifyThread(instanceId, subName string) {
	v := NewInstanceClient()
	iss := NewInstanceStatusSubClient()
	var status = InstanceStatus{
		ResourceCount: -1,
	}
	key := StatusSubscriptionKey{
		InstanceId:       instanceId,
		SubscriptionName: subName,
	}
	time.Sleep(time.Second * 5)
	log.Info("[START] status verification thread", log.Fields{
		"InstanceId":       instanceId,
		"SubscriptionName": subName,
	})

	lastChange := time.Now()
	var timeInSeconds time.Duration = 5
	for {
		time.Sleep(time.Second * timeInSeconds)

		lock, subData, watchers := getSubscriptionData(instanceId)
		var changeDetected = false
		lock.Lock()
		watcherStatus, ok := watchers.watchersStatus[subName]
		if ok {
			changeDetected = watcherStatus.lastUpdateTime.After(lastChange)
		}
		lock.Unlock()
		if !ok {
			break
		}
		if changeDetected || status.ResourceCount < 0 {
			currentSub, err := iss.Get(instanceId, subName)
			if err != nil {
				log.Error("Error getting current status", log.Fields{
					"error":    err.Error(),
					"instance": instanceId})
				break
			}
			if currentSub.MinNotifyInterval > 5 {
				timeInSeconds = time.Duration(currentSub.MinNotifyInterval)
			} else {
				timeInSeconds = 5
			}
			newStatus, err := v.Status(instanceId, false)
			if err != nil {
				log.Error("Error getting current status", log.Fields{
					"error":    err.Error(),
					"instance": instanceId})
				break
			} else {
				if status.ResourceCount >= 0 {
					var delta = statusDelta(status, newStatus)
					if delta.Delta() {
						log.Info("CHANGE DETECTED", log.Fields{
							"Instance":     instanceId,
							"Subscription": subName,
						})
						lastChange = watcherStatus.lastUpdateTime
						for _, res := range delta.Created {
							log.Info("CREATED", log.Fields{
								"Kind": res.GVK.Kind,
								"Name": res.Name,
							})
						}
						for _, res := range delta.Modified {
							log.Info("MODIFIED", log.Fields{
								"Kind": res.GVK.Kind,
								"Name": res.Name,
							})
						}
						for _, res := range delta.Deleted {
							log.Info("DELETED", log.Fields{
								"Kind": res.GVK.Kind,
								"Name": res.Name,
							})
						}
						// Acquire Mutex
						lock.Lock()
						currentSub.LastUpdateTime = time.Now()
						var notifyResultCh = make(chan notifyResult)
						var newData = notifyChannelData{
							instanceId:   instanceId,
							subscription: currentSub,
							action:       "NOTIFY",
							delta:        delta,
							notifyResult: notifyResultCh,
						}
						subData <- newData
						var notifyResult notifyResult = <-notifyResultCh
						log.Info("Notification sent", log.Fields{
							"InstanceId":       instanceId,
							"SubscriptionName": subName,
							"Result":           notifyResult.result,
						})
						currentSub.LastNotifyStatus = notifyResult.result
						currentSub.LastNotifyTime = notifyResult.time
						err = db.DBconn.Update(iss.storeName, key, iss.tagInst, currentSub)
						if err != nil {
							log.Error("Error updating subscription status", log.Fields{
								"error":    err.Error(),
								"instance": instanceId})
						}
						lock.Unlock()
					}
				}

				status = newStatus
			}
		}
	}
	log.Info("[STOP] status verification thread", log.Fields{
		"InstanceId":       instanceId,
		"SubscriptionName": subName,
	})
}

func statusDelta(first, second InstanceStatus) resourceStatusDelta {
	var delta resourceStatusDelta = resourceStatusDelta{
		Created:  make([]ResourceStatus, 0),
		Deleted:  make([]ResourceStatus, 0),
		Modified: make([]ResourceStatus, 0),
	}
	var firstResList map[string]ResourceStatus = make(map[string]ResourceStatus)
	for _, res := range first.ResourcesStatus {
		firstResList[res.Key()] = res
	}
	for _, res := range second.ResourcesStatus {
		var key string = res.Key()
		if prevRes, ok := firstResList[key]; ok {
			if prevRes.Value() != res.Value() {
				delta.Modified = append(delta.Modified, res)
			}
			delete(firstResList, res.Key())
		} else {
			delta.Created = append(delta.Created, res)
		}
	}
	for _, res := range firstResList {
		delta.Deleted = append(delta.Deleted, res)
	}
	return delta
}