github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-09-08 14:01:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add available protocols to the stats endpoint (colors for methods are coming from server) (#1184)

Browse Source 
* add protocols array to the endpoint

* no message

* no message

* fix tests and small fix for the iteration

* fix the color of the protocol

* Get protocols list and method colors from server

* fix tests

* cr fixes

Co-authored-by: Amit Fainholts <amit@up9.com>
This commit is contained in:
gadotroee
2022-07-05 15:30:39 +03:00
committed by GitHub
parent 52c9251c00
commit ec11b21b51
6 changed files with 115 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
agent/pkg/providers/stats_provider.go
View File

@@ -1,6 +1,9 @@
package providers
import (
"crypto/md5"
"encoding/hex"
"fmt"
"reflect"
"strings"
"sync"
@@ -36,13 +39,13 @@ type SizeAndEntriesCount struct {
type AccumulativeStatsCounter struct {
Name string `json:"name"`
Color string `json:"color"`
EntriesCount int `json:"entriesCount"`
VolumeSizeBytes int `json:"volumeSizeBytes"`
}
type AccumulativeStatsProtocol struct {
AccumulativeStatsCounter
Color string `json:"color"`
Methods []*AccumulativeStatsCounter `json:"methods"`
}
@@ -52,6 +55,7 @@ type AccumulativeStatsProtocolTime struct {
}
type TrafficStatsResponse struct {
Protocols []string `json:"protocols"`
PieStats []*AccumulativeStatsProtocol `json:"pie"`
TimelineStats []*AccumulativeStatsProtocolTime `json:"timeline"`
}
@@ -78,20 +82,36 @@ func GetGeneralStats() *GeneralStats {
func InitProtocolToColor(protocolMap map[string]*api.Protocol) {
for item, value := range protocolMap {
protocolToColor[strings.Split(item, "/")[2]] = value.BackgroundColor
splitted := strings.SplitN(item, "/", 3)
protocolToColor[splitted[len(splitted)-1]] = value.BackgroundColor
}
}
func GetTrafficStats() *TrafficStatsResponse {
bucketsStatsCopy := getBucketStatsCopy()
interval := calculateInterval(bucketsStatsCopy[0].BucketTime.Unix(), bucketsStatsCopy[len(bucketsStatsCopy)-1].BucketTime.Unix()) // in seconds
return &TrafficStatsResponse{
Protocols: getAvailableProtocols(bucketsStatsCopy),
PieStats: getAccumulativeStats(bucketsStatsCopy),
TimelineStats: getAccumulativeStatsTiming(bucketsStatsCopy, interval),
TimelineStats: getAccumulativeStatsTiming(bucketsStatsCopy),
}
}
func EntryAdded(size int, summery *api.BaseEntry) {
generalStats.EntriesCount++
generalStats.EntriesVolumeInGB += float64(size) / (1 << 30)
currentTimestamp := int(time.Now().Unix())
if reflect.Value.IsZero(reflect.ValueOf(generalStats.FirstEntryTimestamp)) {
generalStats.FirstEntryTimestamp = currentTimestamp
}
addToBucketStats(size, summery)
generalStats.LastEntryTimestamp = currentTimestamp
}
func calculateInterval(firstTimestamp int64, lastTimestamp int64) time.Duration {
validDurations := []time.Duration{
time.Minute,
@@ -140,31 +160,17 @@ func getAccumulativeStats(stats BucketStats) []*AccumulativeStatsProtocol {
return convertAccumulativeStatsDictToArray(methodsPerProtocolAggregated)
}
func getAccumulativeStatsTiming(stats BucketStats, interval time.Duration) []*AccumulativeStatsProtocolTime {
func getAccumulativeStatsTiming(stats BucketStats) []*AccumulativeStatsProtocolTime {
if len(stats) == 0 {
return make([]*AccumulativeStatsProtocolTime, 0)
}
methodsPerProtocolPerTimeAggregated := getAggregatedResultTiming(interval, stats)
interval := calculateInterval(stats[0].BucketTime.Unix(), stats[len(stats)-1].BucketTime.Unix()) // in seconds
methodsPerProtocolPerTimeAggregated := getAggregatedResultTiming(stats, interval)
return convertAccumulativeStatsTimelineDictToArray(methodsPerProtocolPerTimeAggregated)
}
func EntryAdded(size int, summery *api.BaseEntry) {
generalStats.EntriesCount++
generalStats.EntriesVolumeInGB += float64(size) / (1 << 30)
currentTimestamp := int(time.Now().Unix())
if reflect.Value.IsZero(reflect.ValueOf(generalStats.FirstEntryTimestamp)) {
generalStats.FirstEntryTimestamp = currentTimestamp
}
addToBucketStats(size, summery)
generalStats.LastEntryTimestamp = currentTimestamp
}
func addToBucketStats(size int, summery *api.BaseEntry) {
entryTimeBucketRounded := getBucketFromTimeStamp(summery.Timestamp)
@@ -207,11 +213,11 @@ func convertAccumulativeStatsTimelineDictToArray(methodsPerProtocolPerTimeAggreg
finalResult := make([]*AccumulativeStatsProtocolTime, 0)
for timeKey, item := range methodsPerProtocolPerTimeAggregated {
protocolsData := make([]*AccumulativeStatsProtocol, 0)
for protocolName := range item {
for protocolName, value := range item {
entriesCount := 0
volumeSizeBytes := 0
methods := make([]*AccumulativeStatsCounter, 0)
for _, methodAccData := range methodsPerProtocolPerTimeAggregated[timeKey][protocolName] {
for _, methodAccData := range value {
entriesCount += methodAccData.EntriesCount
volumeSizeBytes += methodAccData.VolumeSizeBytes
methods = append(methods, methodAccData)
@@ -219,10 +225,10 @@ func convertAccumulativeStatsTimelineDictToArray(methodsPerProtocolPerTimeAggreg
protocolsData = append(protocolsData, &AccumulativeStatsProtocol{
AccumulativeStatsCounter: AccumulativeStatsCounter{
Name: protocolName,
Color: protocolToColor[protocolName],
EntriesCount: entriesCount,
VolumeSizeBytes: volumeSizeBytes,
},
Color: protocolToColor[protocolName],
Methods: methods,
})
}
@@ -248,10 +254,10 @@ func convertAccumulativeStatsDictToArray(methodsPerProtocolAggregated map[string
protocolsData = append(protocolsData, &AccumulativeStatsProtocol{
AccumulativeStatsCounter: AccumulativeStatsCounter{
Name: protocolName,
Color: protocolToColor[protocolName],
EntriesCount: entriesCount,
VolumeSizeBytes: volumeSizeBytes,
},
Color: protocolToColor[protocolName],
Methods: methods,
})
}
@@ -269,7 +275,7 @@ func getBucketStatsCopy() BucketStats {
return bucketStatsCopy
}
func getAggregatedResultTiming(interval time.Duration, stats BucketStats) map[time.Time]map[string]map[string]*AccumulativeStatsCounter {
func getAggregatedResultTiming(stats BucketStats, interval time.Duration) map[time.Time]map[string]map[string]*AccumulativeStatsCounter {
methodsPerProtocolPerTimeAggregated := map[time.Time]map[string]map[string]*AccumulativeStatsCounter{}
bucketStatsIndex := len(stats) - 1
@@ -289,6 +295,7 @@ func getAggregatedResultTiming(interval time.Duration, stats BucketStats) map[ti
if _, ok := methodsPerProtocolPerTimeAggregated[resultBucketRoundedKey][protocolName][methodName]; !ok {
methodsPerProtocolPerTimeAggregated[resultBucketRoundedKey][protocolName][methodName] = &AccumulativeStatsCounter{
Name: methodName,
Color: getColorForMethod(protocolName, methodName),
EntriesCount: 0,
VolumeSizeBytes: 0,
}
@@ -303,9 +310,9 @@ func getAggregatedResultTiming(interval time.Duration, stats BucketStats) map[ti
return methodsPerProtocolPerTimeAggregated
}
func getAggregatedStats(bucketStatsCopy BucketStats) map[string]map[string]*AccumulativeStatsCounter {
func getAggregatedStats(stats BucketStats) map[string]map[string]*AccumulativeStatsCounter {
methodsPerProtocolAggregated := make(map[string]map[string]*AccumulativeStatsCounter, 0)
for _, countersOfTimeFrame := range bucketStatsCopy {
for _, countersOfTimeFrame := range stats {
for protocolName, value := range countersOfTimeFrame.ProtocolStats {
for method, countersValue := range value.MethodsStats {
if _, found := methodsPerProtocolAggregated[protocolName]; !found {
@@ -314,6 +321,7 @@ func getAggregatedStats(bucketStatsCopy BucketStats) map[string]map[string]*Accu
if _, found := methodsPerProtocolAggregated[protocolName][method]; !found {
methodsPerProtocolAggregated[protocolName][method] = &AccumulativeStatsCounter{
Name: method,
Color: getColorForMethod(protocolName, method),
EntriesCount: 0,
VolumeSizeBytes: 0,
}
@@ -325,3 +333,25 @@ func getAggregatedStats(bucketStatsCopy BucketStats) map[string]map[string]*Accu
}
return methodsPerProtocolAggregated
}
func getColorForMethod(protocolName string, methodName string) string {
hash := md5.Sum([]byte(fmt.Sprintf("%v_%v", protocolName, methodName)))
input := hex.EncodeToString(hash[:])
return fmt.Sprintf("#%v", input[:6])
}
func getAvailableProtocols(stats BucketStats) []string {
protocols := map[string]bool{}
for _, countersOfTimeFrame := range stats {
for protocolName := range countersOfTimeFrame.ProtocolStats {
protocols[protocolName] = true
}
}
result := make([]string, 0)
for protocol := range protocols {
result = append(result, protocol)
}
result = append(result, "ALL")
return result
}