event.go

/*
 * Copyright 2021 CloudWeGo Authors
 *
 * 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 stats

import (
	"errors"
	"sync"
	"sync/atomic"
)

// EventIndex indicates a unique event.
type EventIndex int

// Level sets the record level.
type Level int

// Event levels.
const (
	LevelDisabled Level = iota
	LevelBase
	LevelDetailed
)

// Event is used to indicate a specific event.
type Event interface {
	Index() EventIndex
	Level() Level
}

type event struct {
	idx   EventIndex
	level Level
}

// Index implements the Event interface.
func (e event) Index() EventIndex {
	return e.idx
}

// Level implements the Event interface.
func (e event) Level() Level {
	return e.level
}

const (
	_ EventIndex = iota
	serverHandleStart
	serverHandleFinish
	clientConnStart
	clientConnFinish
	rpcStart
	rpcFinish
	readStart
	readFinish
	waitReadStart
	waitReadFinish
	writeStart
	writeFinish
	streamRecv
	streamSend

	// NOTE: add new events before this line
	predefinedEventNum
)

// Predefined events.
var (
	RPCStart  = newEvent(rpcStart, LevelBase)
	RPCFinish = newEvent(rpcFinish, LevelBase)

	ServerHandleStart  = newEvent(serverHandleStart, LevelDetailed)
	ServerHandleFinish = newEvent(serverHandleFinish, LevelDetailed)
	ClientConnStart    = newEvent(clientConnStart, LevelDetailed)
	ClientConnFinish   = newEvent(clientConnFinish, LevelDetailed)
	ReadStart          = newEvent(readStart, LevelDetailed)
	ReadFinish         = newEvent(readFinish, LevelDetailed)
	WaitReadStart      = newEvent(waitReadStart, LevelDetailed)
	WaitReadFinish     = newEvent(waitReadFinish, LevelDetailed)
	WriteStart         = newEvent(writeStart, LevelDetailed)
	WriteFinish        = newEvent(writeFinish, LevelDetailed)

	// Streaming Events
	StreamRecv = newEvent(streamRecv, LevelDetailed)
	StreamSend = newEvent(streamSend, LevelDetailed)
)

// errors
var (
	ErrNotAllowed = errors.New("event definition is not allowed after initialization")
	ErrDuplicated = errors.New("event name is already defined")
)

var (
	lock        sync.RWMutex
	inited      int32
	userDefined = make(map[string]Event)
	maxEventNum = int(predefinedEventNum)
)

// FinishInitialization freezes all events defined and prevents further definitions to be added.
func FinishInitialization() {
	lock.Lock()
	defer lock.Unlock()
	atomic.StoreInt32(&inited, 1)
}

// DefineNewEvent allows user to add event definitions during program initialization.
func DefineNewEvent(name string, level Level) (Event, error) {
	if atomic.LoadInt32(&inited) == 1 {
		return nil, ErrNotAllowed
	}
	lock.Lock()
	defer lock.Unlock()
	evt, exist := userDefined[name]
	if exist {
		return evt, ErrDuplicated
	}
	userDefined[name] = newEvent(EventIndex(maxEventNum), level)
	maxEventNum++
	return userDefined[name], nil
}

// MaxEventNum returns the number of event defined.
func MaxEventNum() int {
	lock.RLock()
	defer lock.RUnlock()
	return maxEventNum
}

// PredefinedEventNum returns the number of predefined events of kitex.
func PredefinedEventNum() int {
	return int(predefinedEventNum)
}

func newEvent(idx EventIndex, level Level) Event {
	return event{
		idx:   idx,
		level: level,
	}
}