github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-07-23 20:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement the representations for Kafka Produce

Browse Source
This commit is contained in:
M. Mert Yildiran 2021-08-22 16:14:13 +03:00
parent 89eb0e0f9a
commit 3d16ccf055
No known key found for this signature in database
GPG Key ID: D42ADB236521BF7A
2 changed files with 77 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
tap/extensions/kafka/helpers.go
View File

@ -79,10 +79,14 @@ func representMetadataRequest(data map[string]interface{}) []interface{} {
rep = representRequestHeader(data, rep)
payload := data["Payload"].(map[string]interface{})
topics, _ := json.Marshal(payload["Topics"].([]interface{}))
topics := ""
allowAutoTopicCreation := ""
includeClusterAuthorizedOperations := ""
includeTopicAuthorizedOperations := ""
if payload["Topics"] != nil {
x, _ := json.Marshal(payload["Topics"].([]interface{}))
topics = string(x)
}
if payload["AllowAutoTopicCreation"] != nil {
allowAutoTopicCreation = strconv.FormatBool(payload["AllowAutoTopicCreation"].(bool))
}
@ -95,7 +99,7 @@ func representMetadataRequest(data map[string]interface{}) []interface{} {
repPayload, _ := json.Marshal([]map[string]string{
{
"name": "Topics",
"value": string(topics),
"value": topics,
},
{
"name": "Allow Auto Topic Creation",
@ -244,3 +248,71 @@ func representApiVersionsResponse(data map[string]interface{}) []interface{} {
return rep
}
func representProduceRequest(data map[string]interface{}) []interface{} {
rep := make([]interface{}, 0)
rep = representRequestHeader(data, rep)
payload := data["Payload"].(map[string]interface{})
topicData, _ := json.Marshal(payload["TopicData"].([]interface{}))
transactionalID := ""
if payload["TransactionalID"] != nil {
transactionalID = payload["TransactionalID"].(string)
}
repPayload, _ := json.Marshal([]map[string]string{
{
"name": "Transactional ID",
"value": transactionalID,
},
{
"name": "Required Acknowledgements",
"value": fmt.Sprintf("%d", int(payload["RequiredAcks"].(float64))),
},
{
"name": "Timeout",
"value": fmt.Sprintf("%d", int(payload["Timeout"].(float64))),
},
{
"name": "Topic Data",
"value": string(topicData),
},
})
rep = append(rep, map[string]string{
"type": "table",
"title": "Payload",
"data": string(repPayload),
})
return rep
}
func representProduceResponse(data map[string]interface{}) []interface{} {
rep := make([]interface{}, 0)
rep = representResponseHeader(data, rep)
payload := data["Payload"].(map[string]interface{})
responses, _ := json.Marshal(payload["Responses"].([]interface{}))
throttleTimeMs := ""
if payload["ThrottleTimeMs"] != nil {
throttleTimeMs = fmt.Sprintf("%d", int(payload["ThrottleTimeMs"].(float64)))
}
repPayload, _ := json.Marshal([]map[string]string{
{
"name": "Responses",
"value": string(responses),
},
{
"name": "Throttle Time (ms)",
"value": throttleTimeMs,
},
})
rep = append(rep, map[string]string{
"type": "table",
"title": "Payload",
"data": string(repPayload),
})
return rep
}

3
tap/extensions/kafka/main.go
View File

@ -180,6 +180,9 @@ func (d dissecting) Represent(entry string) ([]byte, error) {
case ApiVersions:
repRequest = representApiVersionsRequest(reqDetails)
repResponse = representApiVersionsResponse(resDetails)
case Produce:
repRequest = representProduceRequest(reqDetails)
repResponse = representProduceResponse(resDetails)
break
}