<commit_before>// Copyright (c) 2018 Cisco and/or its affiliates.
//
// 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 kvscheduler

import (
	"fmt"
	"runtime/trace"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/ligato/cn-infra/logging"
	kvs "github.com/ligato/vpp-agent/plugins/kvscheduler/api"
	"github.com/ligato/vpp-agent/plugins/kvscheduler/internal/graph"
	"github.com/ligato/vpp-agent/plugins/kvscheduler/internal/utils"
)

// GetTransactionHistory returns history of transactions started within the specified
// time window, or the full recorded history if the timestamps are zero values.
func (s *Scheduler) GetTransactionHistory(since, until time.Time) (history kvs.RecordedTxns) {
	s.historyLock.Lock()
	defer s.historyLock.Unlock()

	if !since.IsZero() && !until.IsZero() && until.Before(since) {
		// invalid time window
		return
	}

	lastBefore := -1
	firstAfter := len(s.txnHistory)

	if !since.IsZero() {
		for ; lastBefore+1 < len(s.txnHistory); lastBefore++ {
			if !s.txnHistory[lastBefore+1].Start.Before(since) {
				break
			}
		}
	}

	if !until.IsZero() {
		for ; firstAfter > 0; firstAfter-- {
			if !s.txnHistory[firstAfter-1].Start.After(until) {
				break
			}
		}
	}

	return s.txnHistory[lastBefore+1 : firstAfter]
}

// GetRecordedTransaction returns record of a transaction referenced by the sequence number.
func (s *Scheduler) GetRecordedTransaction(SeqNum uint64) (txn *kvs.RecordedTxn) {
	s.historyLock.Lock()
	defer s.historyLock.Unlock()

	for _, txn := range s.txnHistory {
		if txn.SeqNum == SeqNum {
			return txn
		}
	}
	return nil
}

// preRecordTxnOp prepares txn operation record - fills attributes that we can even
// before executing the operation.
func (s *Scheduler) preRecordTxnOp(args *applyValueArgs, node graph.Node) *kvs.RecordedTxnOp {
	prevOrigin := getNodeOrigin(node)
	if prevOrigin == kvs.UnknownOrigin {
		// new value
		prevOrigin = args.kv.origin
	}
	_, prevErr := getNodeError(node)
	return &kvs.RecordedTxnOp{
		Key:        args.kv.key,
		PrevValue:  utils.RecordProtoMessage(node.GetValue()),
		NewValue:   utils.RecordProtoMessage(args.kv.value),
		PrevState:  getNodeState(node),
		PrevErr:    prevErr,
		IsDerived:  args.isDerived,
		IsProperty: args.isDerived && s.registry.GetDescriptorForKey(args.kv.key) == nil,
		IsRevert:   args.kv.isRevert,
		IsRetry:    args.isRetry,
		IsRecreate: args.recreating != nil && args.recreating.Has(args.kv.key),
	}
}

// preRecordTransaction logs transaction arguments + plan before execution to
// persist some information in case there is a crash during execution.
func (s *Scheduler) preRecordTransaction(txn *transaction, planned kvs.RecordedTxnOps,
	skippedSimulation bool) *kvs.RecordedTxn {
	defer trace.StartRegion(txn.ctx, "preRecordTransaction").End()

	// allocate new transaction record
	record := &kvs.RecordedTxn{
		PreRecord:      true,
		WithSimulation: !skippedSimulation,
		SeqNum:         txn.seqNum,
		TxnType:        txn.txnType,
		Planned:        planned,
	}
	if txn.txnType == kvs.NBTransaction {
		record.ResyncType = txn.nb.resyncType
		record.Description = txn.nb.description
	}
	if txn.txnType == kvs.RetryFailedOps {
		record.RetryForTxn = txn.retry.txnSeqNum
		record.RetryAttempt = txn.retry.attempt
	}

	// record values sorted alphabetically by keys
	if txn.txnType != kvs.NBTransaction || txn.nb.resyncType != kvs.DownstreamResync {
		for _, kv := range txn.values {
			record.Values = append(record.Values, kvs.RecordedKVPair{
				Key:    kv.key,
				Value:  utils.RecordProtoMessage(kv.value),
				Origin: kv.origin,
			})
		}
		sort.Slice(record.Values, func(i, j int) bool {
			return record.Values[i].Key < record.Values[j].Key
		})
	}

	// send to the log
	if s.Log.GetLevel() >= logging.DebugLevel {
		// build header for the log
		txnInfo := txn.txnType.String()
		if txn.txnType == kvs.NBTransaction && txn.nb.resyncType != kvs.NotResync {
			resyncType := "Full Resync"
			if txn.nb.resyncType == kvs.DownstreamResync {
				resyncType = "SB Sync"
			}
			if txn.nb.resyncType == kvs.UpstreamResync {
				resyncType = "NB Sync"
			}
			txnInfo = fmt.Sprintf("%s (%s)", txn.txnType.String(), resyncType)
		}
		var buf strings.Builder
		buf.WriteString("+======================================================================================================================+\n")
		msg := fmt.Sprintf("Transaction #%d", record.SeqNum)
		n := 115 - len(msg)
		buf.WriteString(fmt.Sprintf("| %s %"+strconv.Itoa(n)+"s |\n", msg, txnInfo))
		buf.WriteString("+======================================================================================================================+\n")
		buf.WriteString(record.StringWithOpts(false, false, 2))
		fmt.Println(buf.String())
	}

	return record
}

// recordTransaction records the finalized transaction (log + in-memory).
func (s *Scheduler) recordTransaction(txn *transaction, txnRecord *kvs.RecordedTxn, executed kvs.RecordedTxnOps, start, stop time.Time) {
	defer trace.StartRegion(txn.ctx, "recordTransaction").End()

	txnRecord.PreRecord = false
	txnRecord.Start = start
	txnRecord.Stop = stop
	txnRecord.Executed = executed

	if s.Log.GetLevel() >= logging.DebugLevel {
		var buf strings.Builder
		buf.WriteString("o----------------------------------------------------------------------------------------------------------------------o\n")
		buf.WriteString(txnRecord.StringWithOpts(true, false, 2))
		buf.WriteString("x----------------------------------------------------------------------------------------------------------------------x\n")
		msg := fmt.Sprintf("#%d", txnRecord.SeqNum)
		msg2 := fmt.Sprintf("took %v", stop.Sub(start).Round(time.Millisecond))
		buf.WriteString(fmt.Sprintf("| %s %"+fmt.Sprint(115-len(msg))+"s |\n", msg, msg2))<commit_msg>Round tx duration to 0.1ms<commit_after>// Copyright (c) 2018 Cisco and/or its affiliates.
//
// 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 kvscheduler

import (
	"fmt"
	"runtime/trace"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/ligato/cn-infra/logging"
	kvs "github.com/ligato/vpp-agent/plugins/kvscheduler/api"
	"github.com/ligato/vpp-agent/plugins/kvscheduler/internal/graph"
	"github.com/ligato/vpp-agent/plugins/kvscheduler/internal/utils"
)

// GetTransactionHistory returns history of transactions started within the specified
// time window, or the full recorded history if the timestamps are zero values.
func (s *Scheduler) GetTransactionHistory(since, until time.Time) (history kvs.RecordedTxns) {
	s.historyLock.Lock()
	defer s.historyLock.Unlock()

	if !since.IsZero() && !until.IsZero() && until.Before(since) {
		// invalid time window
		return
	}

	lastBefore := -1
	firstAfter := len(s.txnHistory)

	if !since.IsZero() {
		for ; lastBefore+1 < len(s.txnHistory); lastBefore++ {
			if !s.txnHistory[lastBefore+1].Start.Before(since) {
				break
			}
		}
	}

	if !until.IsZero() {
		for ; firstAfter > 0; firstAfter-- {
			if !s.txnHistory[firstAfter-1].Start.After(until) {
				break
			}
		}
	}

	return s.txnHistory[lastBefore+1 : firstAfter]
}

// GetRecordedTransaction returns record of a transaction referenced by the sequence number.
func (s *Scheduler) GetRecordedTransaction(SeqNum uint64) (txn *kvs.RecordedTxn) {
	s.historyLock.Lock()
	defer s.historyLock.Unlock()

	for _, txn := range s.txnHistory {
		if txn.SeqNum == SeqNum {
			return txn
		}
	}
	return nil
}

// preRecordTxnOp prepares txn operation record - fills attributes that we can even
// before executing the operation.
func (s *Scheduler) preRecordTxnOp(args *applyValueArgs, node graph.Node) *kvs.RecordedTxnOp {
	prevOrigin := getNodeOrigin(node)
	if prevOrigin == kvs.UnknownOrigin {
		// new value
		prevOrigin = args.kv.origin
	}
	_, prevErr := getNodeError(node)
	return &kvs.RecordedTxnOp{
		Key:        args.kv.key,
		PrevValue:  utils.RecordProtoMessage(node.GetValue()),
		NewValue:   utils.RecordProtoMessage(args.kv.value),
		PrevState:  getNodeState(node),
		PrevErr:    prevErr,
		IsDerived:  args.isDerived,
		IsProperty: args.isDerived && s.registry.GetDescriptorForKey(args.kv.key) == nil,
		IsRevert:   args.kv.isRevert,
		IsRetry:    args.isRetry,
		IsRecreate: args.recreating != nil && args.recreating.Has(args.kv.key),
	}
}

// preRecordTransaction logs transaction arguments + plan before execution to
// persist some information in case there is a crash during execution.
func (s *Scheduler) preRecordTransaction(txn *transaction, planned kvs.RecordedTxnOps,
	skippedSimulation bool) *kvs.RecordedTxn {
	defer trace.StartRegion(txn.ctx, "preRecordTransaction").End()

	// allocate new transaction record
	record := &kvs.RecordedTxn{
		PreRecord:      true,
		WithSimulation: !skippedSimulation,
		SeqNum:         txn.seqNum,
		TxnType:        txn.txnType,
		Planned:        planned,
	}
	if txn.txnType == kvs.NBTransaction {
		record.ResyncType = txn.nb.resyncType
		record.Description = txn.nb.description
	}
	if txn.txnType == kvs.RetryFailedOps {
		record.RetryForTxn = txn.retry.txnSeqNum
		record.RetryAttempt = txn.retry.attempt
	}

	// record values sorted alphabetically by keys
	if txn.txnType != kvs.NBTransaction || txn.nb.resyncType != kvs.DownstreamResync {
		for _, kv := range txn.values {
			record.Values = append(record.Values, kvs.RecordedKVPair{
				Key:    kv.key,
				Value:  utils.RecordProtoMessage(kv.value),
				Origin: kv.origin,
			})
		}
		sort.Slice(record.Values, func(i, j int) bool {
			return record.Values[i].Key < record.Values[j].Key
		})
	}

	// send to the log
	if s.Log.GetLevel() >= logging.DebugLevel {
		// build header for the log
		txnInfo := txn.txnType.String()
		if txn.txnType == kvs.NBTransaction && txn.nb.resyncType != kvs.NotResync {
			resyncType := "Full Resync"
			if txn.nb.resyncType == kvs.DownstreamResync {
				resyncType = "SB Sync"
			}
			if txn.nb.resyncType == kvs.UpstreamResync {
				resyncType = "NB Sync"
			}
			txnInfo = fmt.Sprintf("%s (%s)", txn.txnType.String(), resyncType)
		}
		var buf strings.Builder
		buf.WriteString("+======================================================================================================================+\n")
		msg := fmt.Sprintf("Transaction #%d", record.SeqNum)
		n := 115 - len(msg)
		buf.WriteString(fmt.Sprintf("| %s %"+strconv.Itoa(n)+"s |\n", msg, txnInfo))
		buf.WriteString("+======================================================================================================================+\n")
		buf.WriteString(record.StringWithOpts(false, false, 2))
		fmt.Println(buf.String())
	}

	return record
}

// recordTransaction records the finalized transaction (log + in-memory).
func (s *Scheduler) recordTransaction(txn *transaction, txnRecord *kvs.RecordedTxn, executed kvs.RecordedTxnOps, start, stop time.Time) {
	defer trace.StartRegion(txn.ctx, "recordTransaction").End()

	txnRecord.PreRecord = false
	txnRecord.Start = start
	txnRecord.Stop = stop
	txnRecord.Executed = executed

	if s.Log.GetLevel() >= logging.DebugLevel {
		var buf strings.Builder
		buf.WriteString("o----------------------------------------------------------------------------------------------------------------------o\n")
		buf.WriteString(txnRecord.StringWithOpts(true, false, 2))
		buf.WriteString("x----------------------------------------------------------------------------------------------------------------------x\n")
		msg := fmt.Sprintf("#%d", txnRecord.SeqNum)
		msg2 := fmt.Sprintf("took %v", stop.Sub(start).Round(time.Microsecond*100))
		buf.WriteString(fmt.Sprintf("| %s %"+fmt.Sprint(115-len(msg))+"s |\n", msg, msg2))
{'fraction_non_alphanumeric': 0.1955881181810942, 'fraction_numerical': 0.004539300788404874, 'mean_word_length': 3.4249471458773786, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3810259', 'n_tokens_mistral': 4322, 'n_tokens_neox': 4019, 'n_words': 941}