fixed naming of latency to response time (#388)

Co-authored-by: Rami Berman <rami.berman@up9.com>

Dockerfile

@@ -2,8 +2,10 @@ FROM node:14-slim AS site-build
 
 WORKDIR /app/ui-build
 
-COPY ui .
+COPY ui/package.json .
+COPY ui/package-lock.json .
 RUN npm i
+COPY ui .
 RUN npm run build
 
 

README.md

@@ -159,6 +159,13 @@ Such validation may test response for specific JSON fields, headers, etc.
 
 Please see [TRAFFIC RULES](docs/POLICY_RULES.md) page for more details and syntax.
 
+### OpenAPI Specification (OAS) Contract Monitoring
+
+An OAS/Swagger file can contain schemas under `parameters` and `responses` fields. With `--contract catalogue.yaml`
+CLI option, you can pass your API description to Mizu and the traffic will automatically be validated
+against the contracts.
+
+Please see [CONTRACT MONITORING](docs/CONTRACT_MONITORING.md) page for more details and syntax.
 
 ## How to Run local UI
 

agent/go.mod

@@ -5,6 +5,7 @@ go 1.16
 require (
 	github.com/djherbis/atime v1.0.0
 	github.com/fsnotify/fsnotify v1.4.9
+	github.com/getkin/kin-openapi v0.76.0
 	github.com/gin-contrib/static v0.0.1
 	github.com/gin-gonic/gin v1.7.2
 	github.com/go-playground/locales v0.13.0
@@ -12,9 +13,9 @@ require (
 	github.com/go-playground/validator/v10 v10.5.0
 	github.com/google/martian v2.1.0+incompatible
 	github.com/gorilla/websocket v1.4.2
+	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7
 	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap v0.0.0
 	github.com/up9inc/mizu/tap/api v0.0.0

agent/go.sum

@@ -68,6 +68,10 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoD
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
+github.com/getkin/kin-openapi v0.76.0 h1:j77zg3Ec+k+r+GA3d8hBoXpAc6KX9TbBPrwQGBIy2sY=
+github.com/getkin/kin-openapi v0.76.0/go.mod h1:660oXbgy5JFMKreazJaQTw7o+X00qeSyhcnluiMv+Xg=
+github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
+github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
 github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
 github.com/gin-contrib/static v0.0.1 h1:JVxuvHPuUfkoul12N7dtQw7KRn/pSMq7Ue1Va9Swm1U=
@@ -75,6 +79,8 @@ github.com/gin-contrib/static v0.0.1/go.mod h1:CSxeF+wep05e0kCOsqWdAWbSszmc31zTI
 github.com/gin-gonic/gin v1.6.3/go.mod h1:75u5sXoLsGZoRN5Sgbi1eraJ4GU3++wFwWzhwvtwp4M=
 github.com/gin-gonic/gin v1.7.2 h1:Tg03T9yM2xa8j6I3Z3oqLaQRSmKvxPd6g/2HJ6zICFA=
 github.com/gin-gonic/gin v1.7.2/go.mod h1:jD2toBW3GZUr5UMcdrwQA10I7RuaFOl/SGeDjXkfUtY=
+github.com/go-errors/errors v1.4.1 h1:IvVlgbzSsaUNudsw5dcXSzF3EWyXTi5XrAdngnuhRyg=
+github.com/go-errors/errors v1.4.1/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -83,10 +89,13 @@ github.com/go-logr/logr v0.4.0 h1:K7/B1jt6fIBQVd4Owv2MqGQClcgf0R266+7C/QjRcLc=
 github.com/go-logr/logr v0.4.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-openapi/jsonpointer v0.19.2/go.mod h1:3akKfEdA7DF1sugOqz1dVQHBcuDBPKZGEoHC/NkiQRg=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
+github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY=
+github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonreference v0.19.2/go.mod h1:jMjeRr2HHw6nAVajTXJ4eiUwohSTlpa0o73RUL1owJc=
 github.com/go-openapi/jsonreference v0.19.3/go.mod h1:rjx6GuL8TTa9VaixXglHmQmIL98+wF9xc8zWvFonSJ8=
 github.com/go-openapi/spec v0.19.3/go.mod h1:FpwSN1ksY1eteniUU7X0N/BgJ7a4WvBFVA8Lj9mJglo=
 github.com/go-openapi/swag v0.19.2/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
+github.com/go-openapi/swag v0.19.5 h1:lTz6Ys4CmqqCQmZPBlbQENR1/GucA2bzYTE12Pw4tFY=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-playground/assert/v2 v2.0.1 h1:MsBgLAaY856+nPRTKrp3/OZK38U/wa0CcBYNjji3q3A=
 github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
@@ -178,6 +187,8 @@ github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gnostic v0.4.1 h1:DLJCy1n/vrD4HPjOvYcT8aYQXpPIzoRZONaYwyycI+I=
 github.com/googleapis/gnostic v0.4.1/go.mod h1:LRhVm6pbyptWbWbuZ38d1eyptfvIytN3ir6b65WBswg=
+github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
+github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
@@ -216,6 +227,7 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/leodido/go-urn v1.2.0 h1:hpXL4XnriNwQ/ABnpepYM/1vCLWNDfUNts8dX3xTG6Y=
 github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
 github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
+github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e h1:hB2xlXdHp/pmPZq0y3QnmWAArdw9PqbmotexnWx/FU8=
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/markbates/oncer v0.0.0-20181203154359-bf2de49a0be2/go.mod h1:Ld9puTsIW75CHf65OeIOkyKbteujpZVXDpWK6YGZbxE=
 github.com/markbates/safe v1.0.1/go.mod h1:nAqgmRi7cY2nqMc92/bSEeQA+R4OheNU2T1kNSCBdG0=
@@ -258,8 +270,6 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
-github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 h1:jkvpcEatpwuMF5O5LVxTnehj6YZ/aEZN4NWD/Xml4pI=
-github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7/go.mod h1:KTrHyWpO1sevuXPZwyeZc72ddWRFqNSKDFl7uVWKpg0=
 github.com/sirupsen/logrus v1.4.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
@@ -275,6 +285,7 @@ github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoH
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
@@ -533,6 +544,7 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

agent/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/gin-contrib/static"
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
+	"github.com/op/go-logging"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap"
@@ -40,7 +41,8 @@ var extensions []*tapApi.Extension             // global
 var extensionsMap map[string]*tapApi.Extension // global
 
 func main() {
-	logger.InitLoggerStderrOnly()
+	logLevel := determineLogLevel()
+	logger.InitLoggerStderrOnly(logLevel)
 	flag.Parse()
 	loadExtensions()
 
@@ -303,3 +305,11 @@ func getSyncEntriesConfig() *shared.SyncEntriesConfig {
 
 	return syncEntriesConfig
 }
+
+func determineLogLevel() (logLevel logging.Level) {
+	logLevel = logging.INFO
+	if os.Getenv(shared.DebugModeEnvVar) == "1" {
+		logLevel = logging.DEBUG
+	}
+	return
+}

agent/pkg/api/contract_validation.go

@@ -0,0 +1,110 @@
+package api
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/getkin/kin-openapi/openapi3"
+	"github.com/getkin/kin-openapi/openapi3filter"
+	"github.com/getkin/kin-openapi/routers"
+	legacyrouter "github.com/getkin/kin-openapi/routers/legacy"
+
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/api"
+)
+
+const (
+	ContractNotApplicable api.ContractStatus = 0
+	ContractPassed        api.ContractStatus = 1
+	ContractFailed        api.ContractStatus = 2
+)
+
+func loadOAS(ctx context.Context) (doc *openapi3.T, contractContent string, router routers.Router, err error) {
+	path := fmt.Sprintf("%s/%s", shared.RulePolicyPath, shared.ContractFileName)
+	bytes, err := ioutil.ReadFile(path)
+	if err != nil {
+		logger.Log.Error(err.Error())
+		return
+	}
+	contractContent = string(bytes)
+	loader := &openapi3.Loader{Context: ctx}
+	doc, _ = loader.LoadFromData(bytes)
+	err = doc.Validate(ctx)
+	if err != nil {
+		logger.Log.Error(err.Error())
+		return
+	}
+	router, _ = legacyrouter.NewRouter(doc)
+	return
+}
+
+func validateOAS(ctx context.Context, doc *openapi3.T, router routers.Router, req *http.Request, res *http.Response) (isValid bool, reqErr error, resErr error) {
+	isValid = true
+	reqErr = nil
+	resErr = nil
+
+	// Find route
+	route, pathParams, err := router.FindRoute(req)
+	if err != nil {
+		return
+	}
+
+	// Validate request
+	requestValidationInput := &openapi3filter.RequestValidationInput{
+		Request:    req,
+		PathParams: pathParams,
+		Route:      route,
+	}
+	if reqErr = openapi3filter.ValidateRequest(ctx, requestValidationInput); reqErr != nil {
+		isValid = false
+	}
+
+	responseValidationInput := &openapi3filter.ResponseValidationInput{
+		RequestValidationInput: requestValidationInput,
+		Status:                 res.StatusCode,
+		Header:                 res.Header,
+	}
+
+	if res.Body != nil {
+		body, _ := ioutil.ReadAll(res.Body)
+		res.Body = ioutil.NopCloser(bytes.NewBuffer(body))
+		responseValidationInput.SetBodyBytes(body)
+	}
+
+	// Validate response.
+	if resErr = openapi3filter.ValidateResponse(ctx, responseValidationInput); resErr != nil {
+		isValid = false
+	}
+
+	return
+}
+
+func handleOAS(ctx context.Context, doc *openapi3.T, router routers.Router, req *http.Request, res *http.Response, contractContent string) (contract api.Contract) {
+	contract = api.Contract{
+		Content: contractContent,
+		Status:  ContractNotApplicable,
+	}
+
+	isValid, reqErr, resErr := validateOAS(ctx, doc, router, req, res)
+	if isValid {
+		contract.Status = ContractPassed
+	} else {
+		contract.Status = ContractFailed
+		if reqErr != nil {
+			contract.RequestReason = reqErr.Error()
+		} else {
+			contract.RequestReason = ""
+		}
+		if resErr != nil {
+			contract.ResponseReason = resErr.Error()
+		} else {
+			contract.ResponseReason = ""
+		}
+	}
+
+	return
+}

agent/pkg/api/main.go

@@ -99,6 +99,14 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 		panic("Channel of captured messages is nil")
 	}
 
+	disableOASValidation := false
+	ctx := context.Background()
+	doc, contractContent, router, err := loadOAS(ctx)
+	if err != nil {
+		logger.Log.Infof("Disabled OAS validation: %s\n", err.Error())
+		disableOASValidation = true
+	}
+
 	for item := range outputItems {
 		providers.EntryAdded()
 
@@ -107,8 +115,19 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 		mizuEntry := extension.Dissector.Analyze(item, primitive.NewObjectID().Hex(), resolvedSource, resolvedDestionation)
 		baseEntry := extension.Dissector.Summarize(mizuEntry)
 		mizuEntry.EstimatedSizeBytes = getEstimatedEntrySizeBytes(mizuEntry)
-		database.CreateEntry(mizuEntry)
 		if extension.Protocol.Name == "http" {
+			if !disableOASValidation {
+				var httpPair tapApi.HTTPRequestResponsePair
+				json.Unmarshal([]byte(mizuEntry.Entry), &httpPair)
+
+				contract := handleOAS(ctx, doc, router, httpPair.Request.Payload.RawRequest, httpPair.Response.Payload.RawResponse, contractContent)
+				baseEntry.ContractStatus = contract.Status
+				mizuEntry.ContractStatus = contract.Status
+				mizuEntry.ContractRequestReason = contract.RequestReason
+				mizuEntry.ContractResponseReason = contract.ResponseReason
+				mizuEntry.ContractContent = contract.Content
+			}
+
 			var pair tapApi.RequestResponsePair
 			json.Unmarshal([]byte(mizuEntry.Entry), &pair)
 			harEntry, err := utils.NewEntry(&pair)
@@ -117,6 +136,7 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 				baseEntry.Rules = rules
 			}
 		}
+		database.CreateEntry(mizuEntry)
 
 		baseEntryBytes, _ := models.CreateBaseEntryWebSocketMessage(baseEntry)
 		BroadcastToBrowserClients(baseEntryBytes)

agent/pkg/controllers/entries_controller.go

@@ -35,7 +35,6 @@ func GetEntries(c *gin.Context) {
 	database.GetEntriesTable().
 		Order(fmt.Sprintf("timestamp %s", order)).
 		Where(fmt.Sprintf("timestamp %s %v", operatorSymbol, entriesFilter.Timestamp)).
-		Omit("entry"). // remove the "big" entry field
 		Limit(entriesFilter.Limit).
 		Find(&entries)
 
@@ -45,12 +44,21 @@ func GetEntries(c *gin.Context) {
 	}
 
 	baseEntries := make([]tapApi.BaseEntryDetails, 0)
-	for _, data := range entries {
-		harEntry := tapApi.BaseEntryDetails{}
-		if err := models.GetEntry(&data, &harEntry); err != nil {
+	for _, entry := range entries {
+		baseEntryDetails := tapApi.BaseEntryDetails{}
+		if err := models.GetEntry(&entry, &baseEntryDetails); err != nil {
 			continue
 		}
-		baseEntries = append(baseEntries, harEntry)
+
+		var pair tapApi.RequestResponsePair
+		json.Unmarshal([]byte(entry.Entry), &pair)
+		harEntry, err := utils.NewEntry(&pair)
+		if err == nil {
+			rules, _, _ := models.RunValidationRulesState(*harEntry, entry.Service)
+			baseEntryDetails.Rules = rules
+		}
+
+		baseEntries = append(baseEntries, baseEntryDetails)
 	}
 
 	c.JSON(http.StatusOK, baseEntries)

agent/pkg/resolver/resolver.go

@@ -140,6 +140,13 @@ func (resolver *Resolver) watchServices(ctx context.Context) error {
 			serviceHostname := fmt.Sprintf("%s.%s", service.Name, service.Namespace)
 			if service.Spec.ClusterIP != "" && service.Spec.ClusterIP != kubClientNullString {
 				resolver.saveResolvedName(service.Spec.ClusterIP, serviceHostname, event.Type)
+				if service.Spec.Ports != nil {
+					for _, port := range service.Spec.Ports {
+						if port.Port > 0 {
+							resolver.saveResolvedName(fmt.Sprintf("%s:%d", service.Spec.ClusterIP, port.Port), serviceHostname, event.Type)
+						}
+					}
+				}
 				resolver.saveServiceIP(service.Spec.ClusterIP, serviceHostname, event.Type)
 			}
 			if service.Status.LoadBalancer.Ingress != nil {

agent/pkg/rules/rulesHTTP.go

@@ -46,7 +46,7 @@ func ValidateService(serviceFromRule string, service string) bool {
 
 func MatchRequestPolicy(harEntry har.Entry, service string) (resultPolicyToSend []RulesMatched, isEnabled bool) {
 	enforcePolicy, err := shared.DecodeEnforcePolicy(fmt.Sprintf("%s/%s", shared.RulePolicyPath, shared.RulePolicyFileName))
-	if err == nil {
+	if err == nil && len(enforcePolicy.Rules) > 0 {
 		isEnabled = true
 	}
 	for _, rule := range enforcePolicy.Rules {
@@ -111,7 +111,7 @@ func PassedValidationRules(rulesMatched []RulesMatched) (bool, int64, int) {
 		if rule.Matched == false {
 			return false, responseTime, numberOfRulesMatched
 		} else {
-			if strings.ToLower(rule.Rule.Type) == "responseTime" {
+			if strings.ToLower(rule.Rule.Type) == "slo" {
 				if rule.Rule.ResponseTime < responseTime || responseTime == -1 {
 					responseTime = rule.Rule.ResponseTime
 				}

agent/pkg/up9/main.go

@@ -3,6 +3,7 @@ package up9
 import (
 	"bytes"
 	"compress/zlib"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -75,9 +76,6 @@ func getAuthHeader(guestMode bool) string {
 
 func GetTrafficDumpUrl(analyzeDestination string, analyzeModel string) *url.URL {
 	strUrl := fmt.Sprintf("https://traffic.%s/dumpTrafficBulk/%s", analyzeDestination, analyzeModel)
-	if strings.HasPrefix(analyzeDestination, "http") {
-		strUrl = fmt.Sprintf("%s/api/workspace/dumpTrafficBulk", analyzeDestination)
-	}
 	postUrl, _ := url.Parse(strUrl)
 	return postUrl
 }
@@ -132,6 +130,11 @@ func SyncEntries(syncEntriesConfig *shared.SyncEntriesConfig) error {
 		token = fmt.Sprintf("bearer %s", syncEntriesConfig.Token)
 		model = syncEntriesConfig.Workspace
 		guestMode = false
+
+		logger.Log.Infof("Sync entries - upserting model. env %s, model %s\n", syncEntriesConfig.Env, model)
+		if err := upsertModel(token, model, syncEntriesConfig.Env); err != nil {
+			return fmt.Errorf("failed upserting model, err: %v", err)
+		}
 	}
 
 	modelRegex, _ := regexp.Compile("[A-Za-z0-9][-A-Za-z0-9_.]*[A-Za-z0-9]+$")
@@ -145,6 +148,31 @@ func SyncEntries(syncEntriesConfig *shared.SyncEntriesConfig) error {
 	return nil
 }
 
+func upsertModel(token string, model string, envPrefix string) error {
+	upsertModelUrl, _ := url.Parse(fmt.Sprintf("https://trcc.%s/models/%s", envPrefix, model))
+
+	authHeader := getAuthHeader(false)
+	req := &http.Request{
+		Method: http.MethodPost,
+		URL:    upsertModelUrl,
+		Header: map[string][]string{
+			authHeader: {token},
+		},
+	}
+
+	response, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("failed request to upsert model, err: %v", err)
+	}
+
+	// In case the model is not created (not 201) and doesn't exists (not 409)
+	if response.StatusCode != 201 && response.StatusCode != 409 {
+		return fmt.Errorf("failed request to upsert model, status code: %v", response.StatusCode)
+	}
+
+	return nil
+}
+
 func createAnonymousToken(envPrefix string) (*GuestToken, error) {
 	tokenUrl := fmt.Sprintf("https://trcc.%s/anonymous/token", envPrefix)
 	if strings.HasPrefix(envPrefix, "http") {
@@ -204,6 +232,12 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 					harEntry.Request.Headers = append(harEntry.Request.Headers, har.Header{Name: "x-mizu-destination", Value: data.ResolvedDestination})
 					harEntry.Request.URL = utils.SetHostname(harEntry.Request.URL, data.ResolvedDestination)
 				}
+
+				// go's default marshal behavior is to encode []byte fields to base64, python's default unmarshal behavior is to not decode []byte fields from base64
+				if harEntry.Response.Content.Text, err = base64.StdEncoding.DecodeString(string(harEntry.Response.Content.Text)); err != nil {
+					continue
+				}
+
 				result = append(result, *harEntry)
 			}
 

cli/cmd/tap.go

@@ -102,4 +102,5 @@ func init() {
 	tapCmd.Flags().Bool(configStructs.DryRunTapName, defaultTapConfig.DryRun, "Preview of all pods matching the regex, without tapping them")
 	tapCmd.Flags().StringP(configStructs.WorkspaceTapName, "w", defaultTapConfig.Workspace, "Uploads traffic to your UP9 workspace for further analysis (requires auth)")
 	tapCmd.Flags().String(configStructs.EnforcePolicyFile, defaultTapConfig.EnforcePolicyFile, "Yaml file path with policy rules")
+	tapCmd.Flags().String(configStructs.ContractFile, defaultTapConfig.ContractFile, "OAS/Swagger file to validate to monitor the contracts")
 }

cli/cmd/tapRunner.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"io/ioutil"
 	"path"
 	"regexp"
 	"strings"
@@ -12,6 +13,7 @@ import (
 	core "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 
+	"github.com/getkin/kin-openapi/openapi3"
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/config/configStructs"
@@ -58,6 +60,30 @@ func RunMizuTap() {
 		}
 	}
 
+	// Read and validate the OAS file
+	var contract string
+	if config.Config.Tap.ContractFile != "" {
+		bytes, err := ioutil.ReadFile(config.Config.Tap.ContractFile)
+		if err != nil {
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error reading contract file: %v", errormessage.FormatError(err)))
+			return
+		}
+		contract = string(bytes)
+
+		ctx := context.Background()
+		loader := &openapi3.Loader{Context: ctx}
+		doc, err := loader.LoadFromData(bytes)
+		if err != nil {
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error loading contract file: %v", errormessage.FormatError(err)))
+			return
+		}
+		err = doc.Validate(ctx)
+		if err != nil {
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error validating contract file: %v", errormessage.FormatError(err)))
+			return
+		}
+	}
+
 	kubernetesProvider, err := kubernetes.NewProvider(config.Config.KubeConfigPath())
 	if err != nil {
 		logger.Log.Error(err)
@@ -104,7 +130,7 @@ func RunMizuTap() {
 	}
 
 	defer finishMizuExecution(kubernetesProvider)
-	if err := createMizuResources(ctx, kubernetesProvider, mizuValidationRules); err != nil {
+	if err := createMizuResources(ctx, kubernetesProvider, mizuValidationRules, contract); err != nil {
 		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error creating resources: %v", errormessage.FormatError(err)))
 		return
 	}
@@ -126,7 +152,7 @@ func readValidationRules(file string) (string, error) {
 	return string(newContent), nil
 }
 
-func createMizuResources(ctx context.Context, kubernetesProvider *kubernetes.Provider, mizuValidationRules string) error {
+func createMizuResources(ctx context.Context, kubernetesProvider *kubernetes.Provider, mizuValidationRules string, contract string) error {
 	if !config.Config.IsNsRestrictedMode() {
 		if err := createMizuNamespace(ctx, kubernetesProvider); err != nil {
 			return err
@@ -137,15 +163,15 @@ func createMizuResources(ctx context.Context, kubernetesProvider *kubernetes.Pro
 		return err
 	}
 
-	if err := createMizuConfigmap(ctx, kubernetesProvider, mizuValidationRules); err != nil {
+	if err := createMizuConfigmap(ctx, kubernetesProvider, mizuValidationRules, contract); err != nil {
 		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to create resources required for policy validation. Mizu will not validate policy rules. error: %v\n", errormessage.FormatError(err)))
 	}
 
 	return nil
 }
 
-func createMizuConfigmap(ctx context.Context, kubernetesProvider *kubernetes.Provider, data string) error {
-	err := kubernetesProvider.CreateConfigMap(ctx, config.Config.MizuResourcesNamespace, mizu.ConfigMapName, data)
+func createMizuConfigmap(ctx context.Context, kubernetesProvider *kubernetes.Provider, data string, contract string) error {
+	err := kubernetesProvider.CreateConfigMap(ctx, config.Config.MizuResourcesNamespace, mizu.ConfigMapName, data, contract)
 	return err
 }
 

cli/config/configStructs/tapConfig.go

@@ -3,8 +3,9 @@ package configStructs
 import (
 	"errors"
 	"fmt"
-	"github.com/up9inc/mizu/shared/units"
 	"regexp"
+
+	"github.com/up9inc/mizu/shared/units"
 )
 
 const (
@@ -18,6 +19,7 @@ const (
 	DryRunTapName                 = "dry-run"
 	WorkspaceTapName              = "workspace"
 	EnforcePolicyFile             = "traffic-validation-file"
+	ContractFile                  = "contract"
 )
 
 type TapConfig struct {
@@ -34,6 +36,7 @@ type TapConfig struct {
 	DryRun                 bool      `yaml:"dry-run" default:"false"`
 	Workspace              string    `yaml:"workspace"`
 	EnforcePolicyFile      string    `yaml:"traffic-validation-file"`
+	ContractFile           string    `yaml:"contract"`
 	ApiServerResources     Resources `yaml:"api-server-resources"`
 	TapperResources        Resources `yaml:"tapper-resources"`
 }

cli/go.mod

@@ -5,9 +5,9 @@ go 1.16
 require (
 	github.com/creasty/defaults v1.5.1
 	github.com/denisbrodbeck/machineid v1.0.1
+	github.com/getkin/kin-openapi v0.79.0
 	github.com/google/go-github/v37 v37.0.0
 	github.com/google/uuid v1.1.2
-	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/spf13/cobra v1.1.3
 	github.com/spf13/pflag v1.0.5
 	github.com/up9inc/mizu/shared v0.0.0

cli/go.sum

@@ -113,6 +113,9 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoD
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fvbommel/sortorder v1.0.1/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=
+github.com/getkin/kin-openapi v0.79.0 h1:YLZIgIhZLq9z5WFHHIK+oWORRfn6jjwr7qN0xak0xbE=
+github.com/getkin/kin-openapi v0.79.0/go.mod h1:660oXbgy5JFMKreazJaQTw7o+X00qeSyhcnluiMv+Xg=
+github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/globalsign/mgo v0.0.0-20180905125535-1ca0a4f7cbcb/go.mod h1:xkRDCp4j0OGD1HRkm4kmhM+pmpv3AKq5SU7GMg4oO/Q=
 github.com/globalsign/mgo v0.0.0-20181015135952-eeefdecb41b8/go.mod h1:xkRDCp4j0OGD1HRkm4kmhM+pmpv3AKq5SU7GMg4oO/Q=
@@ -140,6 +143,8 @@ github.com/go-openapi/jsonpointer v0.17.0/go.mod h1:cOnomiV+CVVwFLk0A/MExoFMjwds
 github.com/go-openapi/jsonpointer v0.18.0/go.mod h1:cOnomiV+CVVwFLk0A/MExoFMjwdsUdVpsRhURCKh+3M=
 github.com/go-openapi/jsonpointer v0.19.2/go.mod h1:3akKfEdA7DF1sugOqz1dVQHBcuDBPKZGEoHC/NkiQRg=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
+github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY=
+github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonreference v0.17.0/go.mod h1:g4xxGn04lDIRh0GJb5QlpE3HfopLOL6uZrK/VgnsK9I=
 github.com/go-openapi/jsonreference v0.18.0/go.mod h1:g4xxGn04lDIRh0GJb5QlpE3HfopLOL6uZrK/VgnsK9I=
 github.com/go-openapi/jsonreference v0.19.2/go.mod h1:jMjeRr2HHw6nAVajTXJ4eiUwohSTlpa0o73RUL1owJc=
@@ -165,6 +170,7 @@ github.com/go-openapi/strfmt v0.19.5/go.mod h1:eftuHTlB/dI8Uq8JJOyRlieZf+WkkxUuk
 github.com/go-openapi/swag v0.17.0/go.mod h1:AByQ+nYG6gQg71GINrmuDXCPWdL640yX49/kXLo40Tg=
 github.com/go-openapi/swag v0.18.0/go.mod h1:AByQ+nYG6gQg71GINrmuDXCPWdL640yX49/kXLo40Tg=
 github.com/go-openapi/swag v0.19.2/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
+github.com/go-openapi/swag v0.19.5 h1:lTz6Ys4CmqqCQmZPBlbQENR1/GucA2bzYTE12Pw4tFY=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-openapi/validate v0.18.0/go.mod h1:Uh4HdOzKt19xGIGm1qHf/ofbX1YQ4Y+MYsct2VUrAJ4=
 github.com/go-openapi/validate v0.19.2/go.mod h1:1tRCw7m3jtI8eNWEEliiAqUIcBztB2KDnRCRMUi7GTA=
@@ -221,6 +227,7 @@ github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.1.0 h1:Hsa8mG0dQ46ij8Sl2AYJDUv1oA9/d6Vk+3LG99Oe02g=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -238,6 +245,7 @@ github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5m
 github.com/googleapis/gnostic v0.4.1 h1:DLJCy1n/vrD4HPjOvYcT8aYQXpPIzoRZONaYwyycI+I=
 github.com/googleapis/gnostic v0.4.1/go.mod h1:LRhVm6pbyptWbWbuZ38d1eyptfvIytN3ir6b65WBswg=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
+github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
@@ -299,6 +307,7 @@ github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329/go.mod h1:C1wdFJiN
 github.com/mailru/easyjson v0.0.0-20190312143242-1de009706dbe/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
+github.com/mailru/easyjson v0.7.0 h1:aizVhC/NAAcKWb+5QsU1iNOZb4Yws5UO2I+aIprQITM=
 github.com/mailru/easyjson v0.7.0/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
 github.com/markbates/pkger v0.17.1/go.mod h1:0JoVlrol20BSywW79rN3kdFFsE5xYM+rSCQDXbLhiuI=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
@@ -405,6 +414,7 @@ github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoH
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
@@ -692,6 +702,7 @@ gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

cli/kubernetes/provider.go

@@ -16,6 +16,7 @@ import (
 
 	"io"
 
+	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/tap/api"
@@ -195,6 +196,11 @@ func (provider *Provider) CreateMizuApiServerPod(ctx context.Context, opts *ApiS
 
 	port := intstr.FromInt(shared.DefaultApiServerPort)
 
+	debugMode := ""
+	if config.Config.DumpLogs {
+		debugMode = "1"
+	}
+
 	pod := &core.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      opts.PodName,
@@ -223,6 +229,10 @@ func (provider *Provider) CreateMizuApiServerPod(ctx context.Context, opts *ApiS
 							Name:  shared.MaxEntriesDBSizeBytesEnvVar,
 							Value: strconv.FormatInt(opts.MaxEntriesDBSizeBytes, 10),
 						},
+						{
+							Name:  shared.DebugModeEnvVar,
+							Value: debugMode,
+						},
 					},
 					Resources: core.ResourceRequirements{
 						Limits: core.ResourceList{
@@ -475,13 +485,14 @@ func (provider *Provider) handleRemovalError(err error) error {
 	return err
 }
 
-func (provider *Provider) CreateConfigMap(ctx context.Context, namespace string, configMapName string, data string) error {
-	if data == "" {
+func (provider *Provider) CreateConfigMap(ctx context.Context, namespace string, configMapName string, data string, contract string) error {
+	if data == "" && contract == "" {
 		return nil
 	}
 
 	configMapData := make(map[string]string, 0)
 	configMapData[shared.RulePolicyFileName] = data
+	configMapData[shared.ContractFileName] = contract
 	configMap := &core.ConfigMap{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "ConfigMap",
@@ -524,6 +535,11 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 		"--nodefrag",
 	}
 
+	debugMode := ""
+	if config.Config.DumpLogs {
+		debugMode = "1"
+	}
+
 	agentContainer := applyconfcore.Container()
 	agentContainer.WithName(tapperPodName)
 	agentContainer.WithImage(podImage)
@@ -531,6 +547,7 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 	agentContainer.WithSecurityContext(applyconfcore.SecurityContext().WithPrivileged(true))
 	agentContainer.WithCommand(mizuCmd...)
 	agentContainer.WithEnv(
+		applyconfcore.EnvVar().WithName(shared.DebugModeEnvVar).WithValue(debugMode),
 		applyconfcore.EnvVar().WithName(shared.HostModeEnvVar).WithValue("1"),
 		applyconfcore.EnvVar().WithName(shared.TappedAddressesPerNodeDictEnvVar).WithValue(string(nodeToTappedPodIPMapJsonStr)),
 		applyconfcore.EnvVar().WithName(shared.GoGCEnvVar).WithValue("12800"),

cli/kubernetes/watch.go

@@ -3,10 +3,15 @@ package kubernetes
 import (
 	"context"
 	"errors"
+	"fmt"
+	"github.com/up9inc/mizu/shared/debounce"
+	"github.com/up9inc/mizu/shared/logger"
 	"regexp"
 	"sync"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
@@ -16,6 +21,7 @@ func FilteredWatch(ctx context.Context, kubernetesProvider *Provider, targetName
 	removedChan := make(chan *corev1.Pod)
 	errorChan := make(chan error)
 
+
 	var wg sync.WaitGroup
 
 	for _, targetNamespace := range targetNamespaces {
@@ -23,36 +29,33 @@ func FilteredWatch(ctx context.Context, kubernetesProvider *Provider, targetName
 
 		go func(targetNamespace string) {
 			defer wg.Done()
-			watcher := kubernetesProvider.GetPodWatcher(ctx, targetNamespace)
+			watchRestartDebouncer := debounce.NewDebouncer(1 * time.Minute, func() {})
 
 			for {
+				watcher := kubernetesProvider.GetPodWatcher(ctx, targetNamespace)
+				err := startWatchLoop(ctx, watcher, podFilter, addedChan, modifiedChan, removedChan) // blocking
+				watcher.Stop()
+
 				select {
-				case e := <-watcher.ResultChan():
-					if e.Object == nil {
-						errorChan <- errors.New("kubernetes pod watch failed")
-						return
-					}
-
-					pod, ok := e.Object.(*corev1.Pod)
-					if !ok {
-						continue
-					}
-
-					if !podFilter.MatchString(pod.Name) {
-						continue
-					}
-
-					switch e.Type {
-					case watch.Added:
-						addedChan <- pod
-					case watch.Modified:
-						modifiedChan <- pod
-					case watch.Deleted:
-						removedChan <- pod
-					}
-				case <-ctx.Done():
-					watcher.Stop()
+				case <- ctx.Done():
 					return
+				default:
+					break
+				}
+
+				if err != nil {
+					errorChan <- fmt.Errorf("error in k8 watch: %v", err)
+					break
+				} else {
+					if !watchRestartDebouncer.IsOn() {
+						watchRestartDebouncer.SetOn()
+						logger.Log.Debug("k8s watch channel closed, restarting watcher")
+						time.Sleep(time.Second * 5)
+						continue
+					} else {
+						errorChan <- errors.New("k8s watch unstable, closes frequently")
+						break
+					}
 				}
 			}
 		}(targetNamespace)
@@ -69,3 +72,39 @@ func FilteredWatch(ctx context.Context, kubernetesProvider *Provider, targetName
 
 	return addedChan, modifiedChan, removedChan, errorChan
 }
+
+func startWatchLoop(ctx context.Context, watcher watch.Interface, podFilter *regexp.Regexp, addedChan chan *corev1.Pod, modifiedChan chan *corev1.Pod, removedChan chan *corev1.Pod) error {
+	resultChan := watcher.ResultChan()
+	for {
+		select {
+		case e, isChannelOpen := <-resultChan:
+			if !isChannelOpen {
+				return nil
+			}
+
+			if e.Type == watch.Error {
+				return apierrors.FromObject(e.Object)
+			}
+
+			pod, ok := e.Object.(*corev1.Pod)
+			if !ok {
+				continue
+			}
+
+			if !podFilter.MatchString(pod.Name) {
+				continue
+			}
+
+			switch e.Type {
+			case watch.Added:
+				addedChan <- pod
+			case watch.Modified:
+				modifiedChan <- pod
+			case watch.Deleted:
+				removedChan <- pod
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}

docs/CONTRACT_MONITORING.md

@@ -0,0 +1,172 @@
+# OpenAPI Specification (OAS) Contract Monitoring
+
+An OAS/Swagger file can contain schemas under `parameters` and `responses` fields. With `--contract catalogue.yaml`
+CLI option, you can pass your API description to Mizu and the traffic will automatically be validated
+against the contracts.
+
+Below is an example of an OAS/Swagger file from [Sock Shop](https://microservices-demo.github.io/) microservice demo
+that contains a bunch contracts:
+
+```yaml
+openapi: 3.0.1
+info:
+  title: Catalogue resources
+  version: 1.0.0
+  description: ""
+  license:
+    name: MIT
+    url: http://github.com/gruntjs/grunt/blob/master/LICENSE-MIT
+paths:
+  /catalogue:
+    get:
+      description: Catalogue API
+      operationId: List catalogue
+      responses:
+        200:
+          description: ""
+          content:
+            application/json;charset=UTF-8:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/Listresponse'
+  /catalogue/{id}:
+    get:
+      operationId: Get an item
+      parameters:
+      - name: id
+        in: path
+        required: true
+        schema:
+          type: string
+        example: a0a4f044-b040-410d-8ead-4de0446aec7e
+      responses:
+        200:
+          description: ""
+          content:
+            application/json; charset=UTF-8:
+              schema:
+                $ref: '#/components/schemas/Getanitemresponse'
+  /catalogue/size:
+    get:
+      operationId: Get size
+      responses:
+        200:
+          description: ""
+          content:
+            application/json;charset=UTF-8:
+              schema:
+                $ref: '#/components/schemas/Getsizeresponse'
+  /tags:
+    get:
+      operationId: List_
+      responses:
+        200:
+          description: ""
+          content:
+            application/json;charset=UTF-8:
+              schema:
+                $ref: '#/components/schemas/Listresponse3'
+components:
+  schemas:
+    Listresponse:
+      title: List response
+      required:
+      - count
+      - description
+      - id
+      - imageUrl
+      - name
+      - price
+      - tag
+      type: object
+      properties:
+        id:
+          type: string
+        name:
+          type: string
+        description:
+          type: string
+        imageUrl:
+          type: array
+          items:
+            type: string
+        price:
+          type: number
+          format: double
+        count:
+          type: integer
+          format: int32
+        tag:
+          type: array
+          items:
+            type: string
+    Getanitemresponse:
+      title: Get an item response
+      required:
+      - count
+      - description
+      - id
+      - imageUrl
+      - name
+      - price
+      - tag
+      type: object
+      properties:
+        id:
+          type: string
+        name:
+          type: string
+        description:
+          type: string
+        imageUrl:
+          type: array
+          items:
+            type: string
+        price:
+          type: number
+          format: double
+        count:
+          type: integer
+          format: int32
+        tag:
+          type: array
+          items:
+            type: string
+    Getsizeresponse:
+      title: Get size response
+      required:
+      - size
+      type: object
+      properties:
+        size:
+          type: integer
+          format: int32
+    Listresponse3:
+      title: List response3
+      required:
+      - tags
+      type: object
+      properties:
+        tags:
+          type: array
+          items:
+            type: string
+```
+
+Pass it to Mizu through the CLI option: `mizu tap -n sock-shop --contract catalogue.yaml`
+
+Now Mizu will monitor the traffic against these contracts.
+
+If an entry fails to comply with the contract, it's marked with `Breach` notice in the UI.
+The reason of the failure can be seen under the `CONTRACT` tab in the details layout.
+
+### Notes
+
+Make sure that you;
+
+- specified the `openapi` version
+- specified the `info.version` version in the YAML
+- and removed `servers` field from the YAML
+
+Otherwise the OAS file cannot be recognized. (see [this issue](https://github.com/getkin/kin-openapi/issues/356))

shared/consts.go

@@ -9,6 +9,8 @@ const (
 	MaxEntriesDBSizeBytesEnvVar      = "MAX_ENTRIES_DB_BYTES"
 	RulePolicyPath                   = "/app/enforce-policy/"
 	RulePolicyFileName               = "enforce-policy.yaml"
+	ContractFileName                 = "contract-oas.yaml"
 	GoGCEnvVar                       = "GOGC"
 	DefaultApiServerPort             = 8899
+	DebugModeEnvVar                  = "MIZU_DEBUG"
 )

shared/debounce/debounce.go

@@ -52,3 +52,7 @@ func (d *Debouncer) SetOn() error {
 	d.timer = time.AfterFunc(d.timeout, d.callback)
 	return nil
 }
+
+func (d *Debouncer) IsOn() bool {
+	return d.running
+}

shared/logger/logger.go

@@ -29,13 +29,10 @@ func InitLogger(logPath string) {
 	logging.SetBackend(backend1Leveled, backend2Formatter)
 }
 
-func InitLoggerStderrOnly() {
-	consoleLog := logging.NewLogBackend(os.Stderr, "", 0)
+func InitLoggerStderrOnly(level logging.Level) {
+	backend := logging.NewLogBackend(os.Stderr, "", 0)
+	backendFormatter := logging.NewBackendFormatter(backend, format)
 
-	backend1Formatter := logging.NewBackendFormatter(consoleLog, format)
-
-	backend1Leveled := logging.AddModuleLevel(consoleLog)
-	backend1Leveled.SetLevel(logging.DEBUG, "")
-
-	logging.SetBackend(backend1Leveled, backend1Formatter)
+	logging.SetBackend(backendFormatter)
+	logging.SetLevel(level, "")
 }

tap/api/api.go

@@ -2,9 +2,17 @@ package api
 
 import (
 	"bufio"
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
 	"plugin"
 	"sync"
 	"time"
+
+	"github.com/google/martian/har"
 )
 
 type Protocol struct {
@@ -104,32 +112,36 @@ type MizuEntry struct {
 	ID                      uint `gorm:"primarykey"`
 	CreatedAt               time.Time
 	UpdatedAt               time.Time
-	ProtocolName            string `json:"protocolName" gorm:"column:protocolName"`
-	ProtocolLongName        string `json:"protocolLongName" gorm:"column:protocolLongName"`
-	ProtocolAbbreviation    string `json:"protocolAbbreviation" gorm:"column:protocolAbbreviation"`
-	ProtocolVersion         string `json:"protocolVersion" gorm:"column:protocolVersion"`
-	ProtocolBackgroundColor string `json:"protocolBackgroundColor" gorm:"column:protocolBackgroundColor"`
-	ProtocolForegroundColor string `json:"protocolForegroundColor" gorm:"column:protocolForegroundColor"`
-	ProtocolFontSize        int8   `json:"protocolFontSize" gorm:"column:protocolFontSize"`
-	ProtocolReferenceLink   string `json:"protocolReferenceLink" gorm:"column:protocolReferenceLink"`
-	Entry                   string `json:"entry,omitempty" gorm:"column:entry"`
-	EntryId                 string `json:"entryId" gorm:"column:entryId"`
-	Url                     string `json:"url" gorm:"column:url"`
-	Method                  string `json:"method" gorm:"column:method"`
-	Status                  int    `json:"status" gorm:"column:status"`
-	RequestSenderIp         string `json:"requestSenderIp" gorm:"column:requestSenderIp"`
-	Service                 string `json:"service" gorm:"column:service"`
-	Timestamp               int64  `json:"timestamp" gorm:"column:timestamp"`
-	ElapsedTime             int64  `json:"elapsedTime" gorm:"column:elapsedTime"`
-	Path                    string `json:"path" gorm:"column:path"`
-	ResolvedSource          string `json:"resolvedSource,omitempty" gorm:"column:resolvedSource"`
-	ResolvedDestination     string `json:"resolvedDestination,omitempty" gorm:"column:resolvedDestination"`
-	SourceIp                string `json:"sourceIp,omitempty" gorm:"column:sourceIp"`
-	DestinationIp           string `json:"destinationIp,omitempty" gorm:"column:destinationIp"`
-	SourcePort              string `json:"sourcePort,omitempty" gorm:"column:sourcePort"`
-	DestinationPort         string `json:"destinationPort,omitempty" gorm:"column:destinationPort"`
-	IsOutgoing              bool   `json:"isOutgoing,omitempty" gorm:"column:isOutgoing"`
-	EstimatedSizeBytes      int    `json:"-" gorm:"column:estimatedSizeBytes"`
+	ProtocolName            string         `json:"protocolName" gorm:"column:protocolName"`
+	ProtocolLongName        string         `json:"protocolLongName" gorm:"column:protocolLongName"`
+	ProtocolAbbreviation    string         `json:"protocolAbbreviation" gorm:"column:protocolAbbreviation"`
+	ProtocolVersion         string         `json:"protocolVersion" gorm:"column:protocolVersion"`
+	ProtocolBackgroundColor string         `json:"protocolBackgroundColor" gorm:"column:protocolBackgroundColor"`
+	ProtocolForegroundColor string         `json:"protocolForegroundColor" gorm:"column:protocolForegroundColor"`
+	ProtocolFontSize        int8           `json:"protocolFontSize" gorm:"column:protocolFontSize"`
+	ProtocolReferenceLink   string         `json:"protocolReferenceLink" gorm:"column:protocolReferenceLink"`
+	Entry                   string         `json:"entry,omitempty" gorm:"column:entry"`
+	EntryId                 string         `json:"entryId" gorm:"column:entryId"`
+	Url                     string         `json:"url" gorm:"column:url"`
+	Method                  string         `json:"method" gorm:"column:method"`
+	Status                  int            `json:"status" gorm:"column:status"`
+	RequestSenderIp         string         `json:"requestSenderIp" gorm:"column:requestSenderIp"`
+	Service                 string         `json:"service" gorm:"column:service"`
+	Timestamp               int64          `json:"timestamp" gorm:"column:timestamp"`
+	ElapsedTime             int64          `json:"elapsedTime" gorm:"column:elapsedTime"`
+	Path                    string         `json:"path" gorm:"column:path"`
+	ResolvedSource          string         `json:"resolvedSource,omitempty" gorm:"column:resolvedSource"`
+	ResolvedDestination     string         `json:"resolvedDestination,omitempty" gorm:"column:resolvedDestination"`
+	SourceIp                string         `json:"sourceIp,omitempty" gorm:"column:sourceIp"`
+	DestinationIp           string         `json:"destinationIp,omitempty" gorm:"column:destinationIp"`
+	SourcePort              string         `json:"sourcePort,omitempty" gorm:"column:sourcePort"`
+	DestinationPort         string         `json:"destinationPort,omitempty" gorm:"column:destinationPort"`
+	IsOutgoing              bool           `json:"isOutgoing,omitempty" gorm:"column:isOutgoing"`
+	ContractStatus          ContractStatus `json:"contractStatus,omitempty" gorm:"column:contractStatus"`
+	ContractRequestReason   string         `json:"contractRequestReason,omitempty" gorm:"column:contractRequestReason"`
+	ContractResponseReason  string         `json:"contractResponseReason,omitempty" gorm:"column:contractResponseReason"`
+	ContractContent         string         `json:"contractContent,omitempty" gorm:"column:contractContent"`
+	EstimatedSizeBytes      int            `json:"-" gorm:"column:estimatedSizeBytes"`
 }
 
 type MizuEntryWrapper struct {
@@ -159,6 +171,7 @@ type BaseEntryDetails struct {
 	IsOutgoing      bool            `json:"isOutgoing,omitempty"`
 	Latency         int64           `json:"latency"`
 	Rules           ApplicableRules `json:"rules,omitempty"`
+	ContractStatus  ContractStatus  `json:"contractStatus"`
 }
 
 type ApplicableRules struct {
@@ -167,6 +180,15 @@ type ApplicableRules struct {
 	NumberOfRules int   `json:"numberOfRules,omitempty"`
 }
 
+type ContractStatus int
+
+type Contract struct {
+	Status         ContractStatus `json:"status"`
+	RequestReason  string         `json:"requestReason"`
+	ResponseReason string         `json:"responseReason"`
+	Content        string         `json:"content"`
+}
+
 type DataUnmarshaler interface {
 	UnmarshalData(*MizuEntry) error
 }
@@ -184,14 +206,20 @@ func (bed *BaseEntryDetails) UnmarshalData(entry *MizuEntry) error {
 	}
 	bed.Id = entry.EntryId
 	bed.Url = entry.Url
+	bed.RequestSenderIp = entry.RequestSenderIp
 	bed.Service = entry.Service
+	bed.Path = entry.Path
 	bed.Summary = entry.Path
 	bed.StatusCode = entry.Status
 	bed.Method = entry.Method
 	bed.Timestamp = entry.Timestamp
-	bed.RequestSenderIp = entry.RequestSenderIp
+	bed.SourceIp = entry.SourceIp
+	bed.DestinationIp = entry.DestinationIp
+	bed.SourcePort = entry.SourcePort
+	bed.DestinationPort = entry.DestinationPort
 	bed.IsOutgoing = entry.IsOutgoing
 	bed.Latency = entry.ElapsedTime
+	bed.ContractStatus = entry.ContractStatus
 	return nil
 }
 
@@ -199,3 +227,109 @@ const (
 	TABLE string = "table"
 	BODY  string = "body"
 )
+
+const (
+	TypeHttpRequest = iota
+	TypeHttpResponse
+)
+
+type HTTPPayload struct {
+	Type uint8
+	Data interface{}
+}
+
+type HTTPPayloader interface {
+	MarshalJSON() ([]byte, error)
+}
+
+type HTTPWrapper struct {
+	Method      string               `json:"method"`
+	Url         string               `json:"url"`
+	Details     interface{}          `json:"details"`
+	RawRequest  *HTTPRequestWrapper  `json:"rawRequest"`
+	RawResponse *HTTPResponseWrapper `json:"rawResponse"`
+}
+
+func (h HTTPPayload) MarshalJSON() ([]byte, error) {
+	switch h.Type {
+	case TypeHttpRequest:
+		harRequest, err := har.NewRequest(h.Data.(*http.Request), true)
+		if err != nil {
+			return nil, errors.New("Failed converting request to HAR")
+		}
+		return json.Marshal(&HTTPWrapper{
+			Method:     harRequest.Method,
+			Url:        "",
+			Details:    harRequest,
+			RawRequest: &HTTPRequestWrapper{Request: h.Data.(*http.Request)},
+		})
+	case TypeHttpResponse:
+		harResponse, err := har.NewResponse(h.Data.(*http.Response), true)
+		if err != nil {
+			return nil, errors.New("Failed converting response to HAR")
+		}
+		return json.Marshal(&HTTPWrapper{
+			Method:      "",
+			Url:         "",
+			Details:     harResponse,
+			RawResponse: &HTTPResponseWrapper{Response: h.Data.(*http.Response)},
+		})
+	default:
+		panic(fmt.Sprintf("HTTP payload cannot be marshaled: %s\n", h.Type))
+	}
+}
+
+type HTTPWrapperTricky struct {
+	Method      string         `json:"method"`
+	Url         string         `json:"url"`
+	Details     interface{}    `json:"details"`
+	RawRequest  *http.Request  `json:"rawRequest"`
+	RawResponse *http.Response `json:"rawResponse"`
+}
+
+type HTTPMessage struct {
+	IsRequest   bool              `json:"isRequest"`
+	CaptureTime time.Time         `json:"captureTime"`
+	Payload     HTTPWrapperTricky `json:"payload"`
+}
+
+type HTTPRequestResponsePair struct {
+	Request  HTTPMessage `json:"request"`
+	Response HTTPMessage `json:"response"`
+}
+
+type HTTPRequestWrapper struct {
+	*http.Request
+}
+
+func (r *HTTPRequestWrapper) MarshalJSON() ([]byte, error) {
+	body, _ := ioutil.ReadAll(r.Request.Body)
+	r.Request.Body = ioutil.NopCloser(bytes.NewBuffer(body))
+	return json.Marshal(&struct {
+		Body    string `json:"Body,omitempty"`
+		GetBody string `json:"GetBody,omitempty"`
+		Cancel  string `json:"Cancel,omitempty"`
+		*http.Request
+	}{
+		Body:    string(body),
+		Request: r.Request,
+	})
+}
+
+type HTTPResponseWrapper struct {
+	*http.Response
+}
+
+func (r *HTTPResponseWrapper) MarshalJSON() ([]byte, error) {
+	body, _ := ioutil.ReadAll(r.Response.Body)
+	r.Response.Body = ioutil.NopCloser(bytes.NewBuffer(body))
+	return json.Marshal(&struct {
+		Body    string `json:"Body,omitempty"`
+		GetBody string `json:"GetBody,omitempty"`
+		Cancel  string `json:"Cancel,omitempty"`
+		*http.Response
+	}{
+		Body:     string(body),
+		Response: r.Response,
+	})
+}

tap/api/go.mod

@@ -1,3 +1,5 @@
 module github.com/up9inc/mizu/tap/api
 
 go 1.16
+
+require github.com/google/martian v2.1.0+incompatible

tap/api/go.sum

@@ -0,0 +1,2 @@
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
+github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=

tap/cleaner.go

@@ -48,7 +48,7 @@ func (cl *Cleaner) start() {
 	go func() {
 		ticker := time.NewTicker(cl.cleanPeriod)
 
-		for true {
+		for {
 			<-ticker.C
 			cl.clean()
 		}

tap/errors_map.go

@@ -0,0 +1,60 @@
+package tap
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/up9inc/mizu/shared/logger"
+)
+
+type errorsMap struct {
+	errorsMap      map[string]uint
+	outputLevel    int
+	nErrors        uint
+	errorsMapMutex sync.Mutex
+}
+
+func NewErrorsMap(outputLevel int) *errorsMap {
+	return &errorsMap{
+		errorsMap:   make(map[string]uint),
+		outputLevel: outputLevel,
+	}
+}
+
+/* minOutputLevel: Error will be printed only if outputLevel is above this value
+ * t:              key for errorsMap (counting errors)
+ * s, a:           arguments logger.Log.Infof
+ * Note:           Too bad for perf that a... is evaluated
+ */
+func (e *errorsMap) logError(minOutputLevel int, t string, s string, a ...interface{}) {
+	e.errorsMapMutex.Lock()
+	e.nErrors++
+	nb := e.errorsMap[t]
+	e.errorsMap[t] = nb + 1
+	e.errorsMapMutex.Unlock()
+
+	if e.outputLevel >= minOutputLevel {
+		formatStr := fmt.Sprintf("%s: %s", t, s)
+		logger.Log.Errorf(formatStr, a...)
+	}
+}
+
+func (e *errorsMap) Error(t string, s string, a ...interface{}) {
+	e.logError(0, t, s, a...)
+}
+
+func (e *errorsMap) SilentError(t string, s string, a ...interface{}) {
+	e.logError(2, t, s, a...)
+}
+
+func (e *errorsMap) Debug(s string, a ...interface{}) {
+	logger.Log.Debugf(s, a...)
+}
+
+func (e *errorsMap) getErrorsSummary() (int, string) {
+	e.errorsMapMutex.Lock()
+	errorMapLen := len(e.errorsMap)
+	errorsSummery := fmt.Sprintf("%v", e.errorsMap)
+	e.errorsMapMutex.Unlock()
+	return errorMapLen, errorsSummery
+}

tap/extensions/amqp/go.mod

@@ -2,8 +2,6 @@ module github.com/up9inc/mizu/tap/extensions/amqp
 
 go 1.16
 
-require (
-    github.com/up9inc/mizu/tap/api v0.0.0
-)
+require github.com/up9inc/mizu/tap/api v0.0.0
 
 replace github.com/up9inc/mizu/tap/api v0.0.0 => ../../api

tap/extensions/amqp/go.sum

@@ -0,0 +1,2 @@
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
+github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=

tap/extensions/http/go.mod

@@ -3,9 +3,7 @@ module github.com/up9inc/mizu/tap/extensions/http
 go 1.16
 
 require (
-	github.com/beevik/etree v1.1.0 // indirect
-	github.com/google/martian v2.1.0+incompatible
-	github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7
+	github.com/beevik/etree v1.1.0
 	github.com/up9inc/mizu/tap/api v0.0.0
 	golang.org/x/net v0.0.0-20210224082022-3d97a244fca7
 	golang.org/x/text v0.3.5 // indirect

tap/extensions/http/go.sum

@@ -2,8 +2,6 @@ github.com/beevik/etree v1.1.0 h1:T0xke/WvNtMoCqgzPhkX2r4rjY3GDZFi+FjpRZY2Jbs=
 github.com/beevik/etree v1.1.0/go.mod h1:r8Aw8JqVegEf0w2fDnATrX9VpkMcyFeM0FhwO62wh+A=
 github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
-github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 h1:jkvpcEatpwuMF5O5LVxTnehj6YZ/aEZN4NWD/Xml4pI=
-github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7/go.mod h1:KTrHyWpO1sevuXPZwyeZc72ddWRFqNSKDFl7uVWKpg0=
 golang.org/x/net v0.0.0-20210224082022-3d97a244fca7 h1:OgUuv8lsRpBibGNbSizVwKWlysjaNzmC9gYMhPVfqFM=
 golang.org/x/net v0.0.0-20210224082022-3d97a244fca7/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

tap/extensions/http/handlers.go

@@ -8,8 +8,6 @@ import (
 	"io/ioutil"
 	"net/http"
 
-	"github.com/romana/rlog"
-
 	"github.com/up9inc/mizu/tap/api"
 )
 
@@ -85,22 +83,12 @@ func handleHTTP2Stream(grpcAssembler *GrpcAssembler, tcpID *api.TcpID, superTime
 func handleHTTP1ClientStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
 	req, err := http.ReadRequest(b)
 	if err != nil {
-		// log.Println("Error reading stream:", err)
 		return err
 	}
 	counterPair.Request++
 
 	body, err := ioutil.ReadAll(req.Body)
 	req.Body = io.NopCloser(bytes.NewBuffer(body)) // rewind
-	s := len(body)
-	if err != nil {
-		rlog.Debugf("[HTTP-request-body] stream %s Got body err: %s", tcpID.Ident, err)
-	}
-	if err := req.Body.Close(); err != nil {
-		rlog.Debugf("[HTTP-request-body-close] stream %s Failed to close request body: %s", tcpID.Ident, err)
-	}
-	encoding := req.Header["Content-Encoding"]
-	rlog.Tracef(1, "HTTP/1 Request: %s %s %s (Body:%d) -> %s", tcpID.Ident, req.Method, req.URL, s, encoding)
 
 	ident := fmt.Sprintf(
 		"%s->%s %s->%s %d",
@@ -127,30 +115,12 @@ func handleHTTP1ClientStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api
 func handleHTTP1ServerStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
 	res, err := http.ReadResponse(b, nil)
 	if err != nil {
-		// log.Println("Error reading stream:", err)
 		return err
 	}
 	counterPair.Response++
 
 	body, err := ioutil.ReadAll(res.Body)
 	res.Body = io.NopCloser(bytes.NewBuffer(body)) // rewind
-	s := len(body)
-	if err != nil {
-		rlog.Debugf("[HTTP-response-body] HTTP/%s: failed to get body(parsed len:%d): %s", tcpID.Ident, s, err)
-	}
-	if err := res.Body.Close(); err != nil {
-		rlog.Debugf("[HTTP-response-body-close] HTTP/%s: failed to close body(parsed len:%d): %s", tcpID.Ident, s, err)
-	}
-	sym := ","
-	if res.ContentLength > 0 && res.ContentLength != int64(s) {
-		sym = "!="
-	}
-	contentType, ok := res.Header["Content-Type"]
-	if !ok {
-		contentType = []string{http.DetectContentType(body)}
-	}
-	encoding := res.Header["Content-Encoding"]
-	rlog.Tracef(1, "HTTP/1 Response: %s %s (%d%s%d%s) -> %s", tcpID.Ident, res.Status, res.ContentLength, sym, s, contentType, encoding)
 
 	ident := fmt.Sprintf(
 		"%s->%s %s->%s %d",

tap/extensions/http/main.go

@@ -10,8 +10,6 @@ import (
 	"net/url"
 	"time"
 
-	"github.com/romana/rlog"
-
 	"github.com/up9inc/mizu/tap/api"
 )
 
@@ -62,19 +60,11 @@ func (d dissecting) Ping() {
 }
 
 func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
-	ident := fmt.Sprintf("%s->%s:%s->%s", tcpID.SrcIP, tcpID.DstIP, tcpID.SrcPort, tcpID.DstPort)
 	isHTTP2, err := checkIsHTTP2Connection(b, isClient)
-	if err != nil {
-		rlog.Debugf("[HTTP/2-Prepare-Connection] stream %s Failed to check if client is HTTP/2: %s (%v,%+v)", ident, err, err, err)
-		// Do something?
-	}
 
 	var grpcAssembler *GrpcAssembler
 	if isHTTP2 {
-		err := prepareHTTP2Connection(b, isClient)
-		if err != nil {
-			rlog.Debugf("[HTTP/2-Prepare-Connection-After-Check] stream %s error: %s (%v,%+v)", ident, err, err, err)
-		}
+		prepareHTTP2Connection(b, isClient)
 		grpcAssembler = createGrpcAssembler(b)
 	}
 
@@ -89,7 +79,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
-				rlog.Debugf("[HTTP/2] stream %s error: %s (%v,%+v)", ident, err, err, err)
 				continue
 			}
 			dissected = true
@@ -98,7 +87,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
-				rlog.Debugf("[HTTP-request] stream %s Request error: %s (%v,%+v)", ident, err, err, err)
 				continue
 			}
 			dissected = true
@@ -107,7 +95,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
-				rlog.Debugf("[HTTP-response], stream %s Response error: %s (%v,%+v)", ident, err, err, err)
 				continue
 			}
 			dissected = true
@@ -124,7 +111,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 func SetHostname(address, newHostname string) string {
 	replacedUrl, err := url.Parse(address)
 	if err != nil {
-		log.Printf("error replacing hostname to %s in address %s, returning original %v", newHostname, address, err)
 		return address
 	}
 	replacedUrl.Host = newHostname

tap/extensions/http/matcher.go

@@ -7,8 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/romana/rlog"
-
 	"github.com/up9inc/mizu/tap/api"
 )
 
@@ -31,7 +29,7 @@ func (matcher *requestResponseMatcher) registerRequest(ident string, request *ht
 	requestHTTPMessage := api.GenericMessage{
 		IsRequest:   true,
 		CaptureTime: captureTime,
-		Payload: HTTPPayload{
+		Payload: api.HTTPPayload{
 			Type: TypeHttpRequest,
 			Data: request,
 		},
@@ -41,15 +39,12 @@ func (matcher *requestResponseMatcher) registerRequest(ident string, request *ht
 		// Type assertion always succeeds because all of the map's values are of api.GenericMessage type
 		responseHTTPMessage := response.(*api.GenericMessage)
 		if responseHTTPMessage.IsRequest {
-			rlog.Debugf("[Request-Duplicate] Got duplicate request with same identifier")
 			return nil
 		}
-		rlog.Tracef(1, "Matched open Response for %s", key)
 		return matcher.preparePair(&requestHTTPMessage, responseHTTPMessage)
 	}
 
 	matcher.openMessagesMap.Store(key, &requestHTTPMessage)
-	rlog.Tracef(1, "Registered open Request for %s", key)
 	return nil
 }
 
@@ -60,7 +55,7 @@ func (matcher *requestResponseMatcher) registerResponse(ident string, response *
 	responseHTTPMessage := api.GenericMessage{
 		IsRequest:   false,
 		CaptureTime: captureTime,
-		Payload: HTTPPayload{
+		Payload: api.HTTPPayload{
 			Type: TypeHttpResponse,
 			Data: response,
 		},
@@ -70,15 +65,12 @@ func (matcher *requestResponseMatcher) registerResponse(ident string, response *
 		// Type assertion always succeeds because all of the map's values are of api.GenericMessage type
 		requestHTTPMessage := request.(*api.GenericMessage)
 		if !requestHTTPMessage.IsRequest {
-			rlog.Debugf("[Response-Duplicate] Got duplicate response with same identifier")
 			return nil
 		}
-		rlog.Tracef(1, "Matched open Request for %s", key)
 		return matcher.preparePair(requestHTTPMessage, &responseHTTPMessage)
 	}
 
 	matcher.openMessagesMap.Store(key, &responseHTTPMessage)
-	rlog.Tracef(1, "Registered open Response for %s", key)
 	return nil
 }
 

tap/extensions/http/sensitive_data_cleaner.go

@@ -12,7 +12,6 @@ import (
 	"strings"
 
 	"github.com/beevik/etree"
-	"github.com/romana/rlog"
 	"github.com/up9inc/mizu/tap/api"
 )
 
@@ -30,7 +29,7 @@ func IsIgnoredUserAgent(item *api.OutputChannelItem, options *api.TrafficFilteri
 		return false
 	}
 
-	request := item.Pair.Request.Payload.(HTTPPayload).Data.(*http.Request)
+	request := item.Pair.Request.Payload.(api.HTTPPayload).Data.(*http.Request)
 
 	for headerKey, headerValues := range request.Header {
 		if strings.ToLower(headerKey) == "user-agent" {
@@ -50,8 +49,8 @@ func IsIgnoredUserAgent(item *api.OutputChannelItem, options *api.TrafficFilteri
 }
 
 func FilterSensitiveData(item *api.OutputChannelItem, options *api.TrafficFilteringOptions) {
-	request := item.Pair.Request.Payload.(HTTPPayload).Data.(*http.Request)
-	response := item.Pair.Response.Payload.(HTTPPayload).Data.(*http.Response)
+	request := item.Pair.Request.Payload.(api.HTTPPayload).Data.(*http.Request)
+	response := item.Pair.Response.Payload.(api.HTTPPayload).Data.(*http.Response)
 
 	filterHeaders(&request.Header)
 	filterHeaders(&response.Header)
@@ -64,7 +63,6 @@ func filterRequestBody(request *http.Request, options *api.TrafficFilteringOptio
 	contenType := getContentTypeHeaderValue(request.Header)
 	body, err := ioutil.ReadAll(request.Body)
 	if err != nil {
-		rlog.Debugf("Filtering error reading body: %v", err)
 		return
 	}
 	filteredBody, err := filterHttpBody([]byte(body), contenType, options)
@@ -79,7 +77,6 @@ func filterResponseBody(response *http.Response, options *api.TrafficFilteringOp
 	contentType := getContentTypeHeaderValue(response.Header)
 	body, err := ioutil.ReadAll(response.Body)
 	if err != nil {
-		rlog.Debugf("Filtering error reading body: %v", err)
 		return
 	}
 	filteredBody, err := filterHttpBody([]byte(body), contentType, options)

tap/extensions/http/structs.go

@@ -1,55 +0,0 @@
-package main
-
-import (
-	"encoding/json"
-	"errors"
-	"fmt"
-	"net/http"
-
-	"github.com/google/martian/har"
-	"github.com/romana/rlog"
-)
-
-type HTTPPayload struct {
-	Type uint8
-	Data interface{}
-}
-
-type HTTPPayloader interface {
-	MarshalJSON() ([]byte, error)
-}
-
-type HTTPWrapper struct {
-	Method  string      `json:"method"`
-	Url     string      `json:"url"`
-	Details interface{} `json:"details"`
-}
-
-func (h HTTPPayload) MarshalJSON() ([]byte, error) {
-	switch h.Type {
-	case TypeHttpRequest:
-		harRequest, err := har.NewRequest(h.Data.(*http.Request), true)
-		if err != nil {
-			rlog.Debugf("convert-request-to-har", "Failed converting request to HAR %s (%v,%+v)", err, err, err)
-			return nil, errors.New("Failed converting request to HAR")
-		}
-		return json.Marshal(&HTTPWrapper{
-			Method:  harRequest.Method,
-			Url:     "",
-			Details: harRequest,
-		})
-	case TypeHttpResponse:
-		harResponse, err := har.NewResponse(h.Data.(*http.Response), true)
-		if err != nil {
-			rlog.Debugf("convert-response-to-har", "Failed converting response to HAR %s (%v,%+v)", err, err, err)
-			return nil, errors.New("Failed converting response to HAR")
-		}
-		return json.Marshal(&HTTPWrapper{
-			Method:  "",
-			Url:     "",
-			Details: harResponse,
-		})
-	default:
-		panic(fmt.Sprintf("HTTP payload cannot be marshaled: %s\n", h.Type))
-	}
-}

tap/extensions/kafka/go.sum

@@ -7,6 +7,8 @@ github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.5.4 h1:L8R9j+yAqZuZjsqh/z+F1NCffTKKLShY6zXTItVIZ8M=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
+github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/klauspost/compress v1.9.8 h1:VMAMUUOh+gaxKTMk+zqbjsSjsIcUcL/LF4o63i82QyA=
 github.com/klauspost/compress v1.9.8/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=

tap/extensions/redis/go.mod

@@ -2,8 +2,6 @@ module github.com/up9inc/mizu/tap/extensions/redis
 
 go 1.16
 
-require (
-    github.com/up9inc/mizu/tap/api v0.0.0
-)
+require github.com/up9inc/mizu/tap/api v0.0.0
 
 replace github.com/up9inc/mizu/tap/api v0.0.0 => ../../api

tap/extensions/redis/go.sum

@@ -0,0 +1,2 @@
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
+github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=

tap/extensions/redis/read.go

@@ -4,7 +4,6 @@ import (
 	"bufio"
 	"errors"
 	"fmt"
-	"log"
 	"math"
 	"reflect"
 	"strconv"
@@ -315,7 +314,6 @@ func (p *RedisProtocol) Read() (packet *RedisPacket, err error) {
 			}
 		default:
 			msg := fmt.Sprintf("Unrecognized element in Redis array: %v\n", reflect.TypeOf(array[0]))
-			log.Printf(msg)
 			err = errors.New(msg)
 			return
 		}
@@ -336,7 +334,6 @@ func (p *RedisProtocol) Read() (packet *RedisPacket, err error) {
 		packet.Value = fmt.Sprintf("%d", x.(int64))
 	default:
 		msg := fmt.Sprintf("Unrecognized Redis data type: %v\n", reflect.TypeOf(x))
-		log.Printf(msg)
 		err = errors.New(msg)
 		return
 	}

tap/go.mod

@@ -3,15 +3,16 @@ module github.com/up9inc/mizu/tap
 go 1.16
 
 require (
-	github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4 // indirect
+	github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4
+	github.com/go-errors/errors v1.4.1
 	github.com/google/gopacket v1.1.19
-	github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7
+	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
+	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap/api v0.0.0
-	golang.org/x/net v0.0.0-20210224082022-3d97a244fca7
-	golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073
-	golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d
-	golang.org/x/text v0.3.5
-	golang.org/x/tools v0.0.0-20210106214847-113979e3529a
+	golang.org/x/net v0.0.0-20210224082022-3d97a244fca7 // indirect
+	golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073 // indirect
 )
 
 replace github.com/up9inc/mizu/tap/api v0.0.0 => ./api
+
+replace github.com/up9inc/mizu/shared v0.0.0 => ../shared

tap/go.sum

@@ -1,45 +1,34 @@
 github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4 h1:NJOOlc6ZJjix0A1rAU+nxruZtR8KboG1848yqpIUo4M=
 github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4/go.mod h1:DQPxZS994Ld1Y8uwnJT+dRL04XPD0cElP/pHH/zEBHM=
+github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/go-errors/errors v1.4.1 h1:IvVlgbzSsaUNudsw5dcXSzF3EWyXTi5XrAdngnuhRyg=
+github.com/go-errors/errors v1.4.1/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=
+github.com/golang-jwt/jwt/v4 v4.1.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
 github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
 github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
-github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 h1:jkvpcEatpwuMF5O5LVxTnehj6YZ/aEZN4NWD/Xml4pI=
-github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7/go.mod h1:KTrHyWpO1sevuXPZwyeZc72ddWRFqNSKDFl7uVWKpg0=
-github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
+github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
+github.com/op/go-logging v0.0.0-20160315200505-970db520ece7 h1:lDH9UUVJtmYCjyT0CI4q8xvlXPxeZ0gYCVvWbmPlp88=
+github.com/op/go-logging v0.0.0-20160315200505-970db520ece7/go.mod h1:HzydrMdWErDVzsI23lYNej1Htcns9BCg93Dk0bBINWk=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
-golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210224082022-3d97a244fca7 h1:OgUuv8lsRpBibGNbSizVwKWlysjaNzmC9gYMhPVfqFM=
 golang.org/x/net v0.0.0-20210224082022-3d97a244fca7/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d h1:+R4KGOnez64A81RvjARKc4UT5/tI9ujCIVX+P5KiHuI=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073 h1:8qxJSnu+7dRq6upnbntrmriWByIakBuct5OM/MdQC1M=
 golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d h1:SZxvLBoTP5yHO3Frd4z4vrF+DBX9vMVanchswa69toE=
-golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.4 h1:0YWbFKbhXG/wIiuHDSKpS0Iy7FSA+u45VtBMfQcFTTc=
-golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.5 h1:i6eZZ+zk0SOf0xgBpEpPD18qWcJda6q1sxt3S0kzyUQ=
-golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
-golang.org/x/tools v0.0.0-20210106214847-113979e3529a h1:CB3a9Nez8M13wwlr/E2YtwoU+qYHKfC+JrDa45RXXoQ=
-golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

tap/net_utils.go

@@ -27,6 +27,7 @@ func getLocalhostIPs() ([]string, error) {
 	return myIPs, nil
 }
 
+//lint:ignore U1000 will be used in the future
 func isPrivateIP(ipStr string) bool {
 	ip := net.ParseIP(ipStr)
 	if ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() {
@@ -54,7 +55,7 @@ func initPrivateIPBlocks() {
 	} {
 		_, block, err := net.ParseCIDR(cidr)
 		if err != nil {
-			Error("Private-IP-Block-Parse", "parse error on %q: %v", cidr, err)
+			tapErrors.Error("Private-IP-Block-Parse", "parse error on %q: %v", cidr, err)
 		} else {
 			privateIPBlocks = append(privateIPBlocks, block)
 		}

tap/passive_tapper.go

@@ -17,7 +17,6 @@ import (
 	"os"
 	"os/signal"
 	"runtime"
-	_debug "runtime/debug"
 	"runtime/pprof"
 	"strconv"
 	"strings"
@@ -36,6 +35,7 @@ import (
 
 const cleanPeriod = time.Second * 10
 
+//lint:ignore U1000 will be used in the future
 var remoteOnlyOutboundPorts = []int{80, 443}
 
 var maxcount = flag.Int64("c", -1, "Only grab this many packets, then exit")
@@ -63,6 +63,7 @@ var staleTimeoutSeconds = flag.Int("staletimout", 120, "Max time in seconds to k
 var memprofile = flag.String("memprofile", "", "Write memory profile")
 
 var appStats = api.AppStats{}
+var tapErrors *errorsMap
 
 // global
 var stats struct {
@@ -87,47 +88,10 @@ type TapOpts struct {
 	HostMode bool
 }
 
-var outputLevel int
-var errorsMap map[string]uint
-var errorsMapMutex sync.Mutex
-var nErrors uint
-var ownIps []string                               // global
 var hostMode bool                                 // global
 var extensions []*api.Extension                   // global
 var filteringOptions *api.TrafficFilteringOptions // global
 
-const baseStreamChannelTimeoutMs int = 5000 * 100
-
-/* minOutputLevel: Error will be printed only if outputLevel is above this value
- * t:              key for errorsMap (counting errors)
- * s, a:           arguments logger.Log.Infof
- * Note:           Too bad for perf that a... is evaluated
- */
-func logError(minOutputLevel int, t string, s string, a ...interface{}) {
-	errorsMapMutex.Lock()
-	nErrors++
-	nb, _ := errorsMap[t]
-	errorsMap[t] = nb + 1
-	errorsMapMutex.Unlock()
-
-	if outputLevel >= minOutputLevel {
-		formatStr := fmt.Sprintf("%s: %s", t, s)
-		logger.Log.Errorf(formatStr, a...)
-	}
-}
-func Error(t string, s string, a ...interface{}) {
-	logError(0, t, s, a...)
-}
-func SilentError(t string, s string, a ...interface{}) {
-	logError(2, t, s, a...)
-}
-func Debug(s string, a ...interface{}) {
-	logger.Log.Debugf(s, a...)
-}
-func Trace(s string, a ...interface{}) {
-	logger.Log.Infof(s, a...)
-}
-
 func inArrayInt(arr []int, valueToCheck int) bool {
 	for _, value := range arr {
 		if value == valueToCheck {
@@ -194,7 +158,7 @@ func startMemoryProfiler() {
 			}
 		}
 
-		for true {
+		for {
 			t := time.Now()
 
 			filename := fmt.Sprintf("%s/%s__mem.prof", dumpPath, t.Format("15_04_05"))
@@ -215,44 +179,11 @@ func startMemoryProfiler() {
 	}()
 }
 
-func closeTimedoutTcpStreamChannels() {
-	TcpStreamChannelTimeoutMs := GetTcpChannelTimeoutMs()
-	for {
-		time.Sleep(10 * time.Millisecond)
-		_debug.FreeOSMemory()
-		streams.Range(func(key interface{}, value interface{}) bool {
-			streamWrapper := value.(*tcpStreamWrapper)
-			stream := streamWrapper.stream
-			if stream.superIdentifier.Protocol == nil {
-				if !stream.isClosed && time.Now().After(streamWrapper.createdAt.Add(TcpStreamChannelTimeoutMs)) {
-					stream.Close()
-					appStats.IncDroppedTcpStreams()
-					logger.Log.Debugf("Dropped an unidentified TCP stream because of timeout. Total dropped: %d Total Goroutines: %d Timeout (ms): %d\n", appStats.DroppedTcpStreams, runtime.NumGoroutine(), TcpStreamChannelTimeoutMs/1000000)
-				}
-			} else {
-				if !stream.superIdentifier.IsClosedOthers {
-					for i := range stream.clients {
-						reader := &stream.clients[i]
-						if reader.extension.Protocol != stream.superIdentifier.Protocol {
-							reader.Close()
-						}
-					}
-					for i := range stream.servers {
-						reader := &stream.servers[i]
-						if reader.extension.Protocol != stream.superIdentifier.Protocol {
-							reader.Close()
-						}
-					}
-					stream.superIdentifier.IsClosedOthers = true
-				}
-			}
-			return true
-		})
-	}
-}
-
 func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
-	go closeTimedoutTcpStreamChannels()
+	streamsMap := NewTcpStreamMap()
+	go streamsMap.closeTimedoutTcpStreamChannels()
+
+	var outputLevel int
 
 	defer util.Run()()
 	if *debug {
@@ -262,19 +193,12 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 	} else if *quiet {
 		outputLevel = -1
 	}
-	errorsMap = make(map[string]uint)
 
-	if localhostIPs, err := getLocalhostIPs(); err != nil {
-		// TODO: think this over
-		logger.Log.Info("Failed to get self IP addresses")
-		logger.Log.Errorf("Getting-Self-Address", "Error getting self ip address: %s (%v,%+v)", err, err, err)
-		ownIps = make([]string, 0)
-	} else {
-		ownIps = localhostIPs
-	}
+	tapErrors = NewErrorsMap(outputLevel)
 
 	var handle *pcap.Handle
 	var err error
+
 	if *fname != "" {
 		if handle, err = pcap.OpenOffline(*fname); err != nil {
 			logger.Log.Fatalf("PCAP OpenOffline error: %v", err)
@@ -319,7 +243,7 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 	var ok bool
 	decoderName := *decoder
 	if decoderName == "" {
-		decoderName = fmt.Sprintf("%s", handle.LinkType())
+		decoderName = handle.LinkType().String()
 	}
 	if dec, ok = gopacket.DecodersByLayerName[decoderName]; !ok {
 		logger.Log.Fatal("No decoder named", decoderName)
@@ -336,9 +260,7 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 		OutputChannel: outputItems,
 	}
 
-	streamFactory := &tcpStreamFactory{
-		Emitter: emitter,
-	}
+	streamFactory := NewTcpStreamFactory(emitter, streamsMap)
 	streamPool := reassembly.NewStreamPool(streamFactory)
 	assembler := reassembly.NewAssembler(streamPool)
 
@@ -366,17 +288,15 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 		statsPeriod := time.Second * time.Duration(*statsevery)
 		ticker := time.NewTicker(statsPeriod)
 
-		for true {
+		for {
 			<-ticker.C
 
 			// Since the start
-			errorsMapMutex.Lock()
-			errorMapLen := len(errorsMap)
-			errorsSummery := fmt.Sprintf("%v", errorsMap)
-			errorsMapMutex.Unlock()
+			errorMapLen, errorsSummery := tapErrors.getErrorsSummary()
+
 			logger.Log.Infof("%v (errors: %v, errTypes:%v) - Errors Summary: %s",
 				time.Since(appStats.StartTime),
-				nErrors,
+				tapErrors.nErrors,
 				errorMapLen,
 				errorsSummery,
 			)
@@ -413,7 +333,9 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 		if err == io.EOF {
 			break
 		} else if err != nil {
-			logger.Log.Debugf("Error: %v", err)
+			if err.Error() != "Timeout Expired" {
+				logger.Log.Debugf("Error: %T", err)
+			}
 			continue
 		}
 		packetsCount := appStats.IncPacketsCount()
@@ -473,14 +395,12 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 
 		done := *maxcount > 0 && int64(appStats.PacketsCount) >= *maxcount
 		if done {
-			errorsMapMutex.Lock()
-			errorMapLen := len(errorsMap)
-			errorsMapMutex.Unlock()
+			errorMapLen, _ := tapErrors.getErrorsSummary()
 			logger.Log.Infof("Processed %v packets (%v bytes) in %v (errors: %v, errTypes:%v)",
 				appStats.PacketsCount,
 				appStats.ProcessedBytes,
 				time.Since(appStats.StartTime),
-				nErrors,
+				tapErrors.nErrors,
 				errorMapLen)
 		}
 		select {
@@ -534,9 +454,9 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 	logger.Log.Infof(" biggest-chunk bytes:\t%d", stats.biggestChunkBytes)
 	logger.Log.Infof(" overlap packets:\t%d", stats.overlapPackets)
 	logger.Log.Infof(" overlap bytes:\t\t%d", stats.overlapBytes)
-	logger.Log.Infof("Errors: %d", nErrors)
-	for e := range errorsMap {
-		logger.Log.Infof(" %s:\t\t%d", e, errorsMap[e])
+	logger.Log.Infof("Errors: %d", tapErrors.nErrors)
+	for e := range tapErrors.errorsMap {
+		logger.Log.Infof(" %s:\t\t%d", e, tapErrors.errorsMap[e])
 	}
 	logger.Log.Infof("AppStats: %v", GetStats())
 }

tap/tcp_reader.go

@@ -2,7 +2,6 @@ package tap
 
 import (
 	"bufio"
-	"fmt"
 	"io"
 	"io/ioutil"
 	"sync"
@@ -20,13 +19,6 @@ type tcpReaderDataMsg struct {
 	timestamp time.Time
 }
 
-type tcpID struct {
-	srcIP   string
-	dstIP   string
-	srcPort string
-	dstPort string
-}
-
 type ConnectionInfo struct {
 	ClientIP   string
 	ClientPort string
@@ -35,10 +27,6 @@ type ConnectionInfo struct {
 	IsOutgoing bool
 }
 
-func (tid *tcpID) String() string {
-	return fmt.Sprintf("%s->%s %s->%s", tid.srcIP, tid.dstIP, tid.srcPort, tid.dstPort)
-}
-
 /* tcpReader gets reads from a channel of bytes of tcp payload, and parses it into requests and responses.
  * The payload is written to the channel by a tcpStream object that is dedicated to one tcp connection.
  * An tcpReader object is unidirectional: it parses either a client stream or a server stream.
@@ -54,7 +42,6 @@ type tcpReader struct {
 	data               []byte
 	superTimer         *api.SuperTimer
 	parent             *tcpStream
-	messageCount       uint
 	packetsSeen        uint
 	outboundLinkWriter *OutboundLinkWriter
 	extension          *api.Extension

tap/tcp_stream.go

@@ -28,15 +28,15 @@ type tcpStream struct {
 	isTapTarget     bool
 	clients         []tcpReader
 	servers         []tcpReader
-	urls            []string
 	ident           string
 	sync.Mutex
+	streamsMap *tcpStreamMap
 }
 
 func (t *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassembly.TCPFlowDirection, nextSeq reassembly.Sequence, start *bool, ac reassembly.AssemblerContext) bool {
 	// FSM
 	if !t.tcpstate.CheckState(tcp, dir) {
-		SilentError("FSM-rejection", "%s: Packet rejected by FSM (state:%s)", t.ident, t.tcpstate.String())
+		tapErrors.SilentError("FSM-rejection", "%s: Packet rejected by FSM (state:%s)", t.ident, t.tcpstate.String())
 		stats.rejectFsm++
 		if !t.fsmerr {
 			t.fsmerr = true
@@ -49,7 +49,7 @@ func (t *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 	// Options
 	err := t.optchecker.Accept(tcp, ci, dir, nextSeq, start)
 	if err != nil {
-		SilentError("OptionChecker-rejection", "%s: Packet rejected by OptionChecker: %s", t.ident, err)
+		tapErrors.SilentError("OptionChecker-rejection", "%s: Packet rejected by OptionChecker: %s", t.ident, err)
 		stats.rejectOpt++
 		if !*nooptcheck {
 			return false
@@ -60,10 +60,10 @@ func (t *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 	if *checksum {
 		c, err := tcp.ComputeChecksum()
 		if err != nil {
-			SilentError("ChecksumCompute", "%s: Got error computing checksum: %s", t.ident, err)
+			tapErrors.SilentError("ChecksumCompute", "%s: Got error computing checksum: %s", t.ident, err)
 			accept = false
 		} else if c != 0x0 {
-			SilentError("Checksum", "%s: Invalid checksum: 0x%x", t.ident, c)
+			tapErrors.SilentError("Checksum", "%s: Invalid checksum: 0x%x", t.ident, c)
 			accept = false
 		}
 	}
@@ -97,7 +97,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 	if sgStats.OverlapBytes != 0 && sgStats.OverlapPackets == 0 {
 		// In the original example this was handled with panic().
 		// I don't know what this error means or how to handle it properly.
-		SilentError("Invalid-Overlap", "bytes:%d, pkts:%d", sgStats.OverlapBytes, sgStats.OverlapPackets)
+		tapErrors.SilentError("Invalid-Overlap", "bytes:%d, pkts:%d", sgStats.OverlapBytes, sgStats.OverlapPackets)
 	}
 	stats.overlapBytes += sgStats.OverlapBytes
 	stats.overlapPackets += sgStats.OverlapPackets
@@ -108,7 +108,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 	} else {
 		ident = fmt.Sprintf("%v %v(%s): ", t.net.Reverse(), t.transport.Reverse(), dir)
 	}
-	Trace("%s: SG reassembled packet with %d bytes (start:%v,end:%v,skip:%d,saved:%d,nb:%d,%d,overlap:%d,%d)", ident, length, start, end, skip, saved, sgStats.Packets, sgStats.Chunks, sgStats.OverlapBytes, sgStats.OverlapPackets)
+	tapErrors.Debug("%s: SG reassembled packet with %d bytes (start:%v,end:%v,skip:%d,saved:%d,nb:%d,%d,overlap:%d,%d)", ident, length, start, end, skip, saved, sgStats.Packets, sgStats.Chunks, sgStats.OverlapBytes, sgStats.OverlapPackets)
 	if skip == -1 && *allowmissinginit {
 		// this is allowed
 	} else if skip != 0 {
@@ -127,18 +127,18 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		}
 		dnsSize := binary.BigEndian.Uint16(data[:2])
 		missing := int(dnsSize) - len(data[2:])
-		Trace("dnsSize: %d, missing: %d", dnsSize, missing)
+		tapErrors.Debug("dnsSize: %d, missing: %d", dnsSize, missing)
 		if missing > 0 {
-			Debug("Missing some bytes: %d", missing)
+			tapErrors.Debug("Missing some bytes: %d", missing)
 			sg.KeepFrom(0)
 			return
 		}
 		p := gopacket.NewDecodingLayerParser(layers.LayerTypeDNS, dns)
 		err := p.DecodeLayers(data[2:], &decoded)
 		if err != nil {
-			SilentError("DNS-parser", "Failed to decode DNS: %v", err)
+			tapErrors.SilentError("DNS-parser", "Failed to decode DNS: %v", err)
 		} else {
-			Trace("DNS: %s", gopacket.LayerDump(dns))
+			tapErrors.Debug("DNS: %s", gopacket.LayerDump(dns))
 		}
 		if len(data) > 2+int(dnsSize) {
 			sg.KeepFrom(2 + int(dnsSize))
@@ -173,7 +173,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 }
 
 func (t *tcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
-	Trace("%s: Connection closed", t.ident)
+	tapErrors.Debug("%s: Connection closed", t.ident)
 	if t.isTapTarget && !t.isClosed {
 		t.Close()
 	}
@@ -193,7 +193,7 @@ func (t *tcpStream) Close() {
 	if shouldReturn {
 		return
 	}
-	streams.Delete(t.id)
+	t.streamsMap.Delete(t.id)
 
 	for i := range t.clients {
 		reader := &t.clients[i]

tap/tcp_stream_factory.go

@@ -22,6 +22,8 @@ type tcpStreamFactory struct {
 	wg                 sync.WaitGroup
 	outboundLinkWriter *OutboundLinkWriter
 	Emitter            api.Emitter
+	streamsMap         *tcpStreamMap
+	ownIps             []string
 }
 
 type tcpStreamWrapper struct {
@@ -29,8 +31,24 @@ type tcpStreamWrapper struct {
 	createdAt time.Time
 }
 
-var streams *sync.Map = &sync.Map{} // global
-var streamId int64 = 0
+func NewTcpStreamFactory(emitter api.Emitter, streamsMap *tcpStreamMap) *tcpStreamFactory {
+	var ownIps []string
+
+	if localhostIPs, err := getLocalhostIPs(); err != nil {
+		// TODO: think this over
+		logger.Log.Info("Failed to get self IP addresses")
+		logger.Log.Errorf("Getting-Self-Address", "Error getting self ip address: %s (%v,%+v)", err, err, err)
+		ownIps = make([]string, 0)
+	} else {
+		ownIps = localhostIPs
+	}
+
+	return &tcpStreamFactory{
+		Emitter:    emitter,
+		streamsMap: streamsMap,
+		ownIps:     ownIps,
+	}
+}
 
 func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream {
 	logger.Log.Debugf("* NEW: %s %s", net, transport)
@@ -56,10 +74,10 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T
 		ident:           fmt.Sprintf("%s:%s", net, transport),
 		optchecker:      reassembly.NewTCPOptionCheck(),
 		superIdentifier: &api.SuperIdentifier{},
+		streamsMap:      factory.streamsMap,
 	}
 	if stream.isTapTarget {
-		streamId++
-		stream.id = streamId
+		stream.id = factory.streamsMap.nextId()
 		for i, extension := range extensions {
 			counterPair := &api.CounterPair{
 				Request:  0,
@@ -102,7 +120,7 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T
 				counterPair:        counterPair,
 			})
 
-			streams.Store(stream.id, &tcpStreamWrapper{
+			factory.streamsMap.Store(stream.id, &tcpStreamWrapper{
 				stream:    stream,
 				createdAt: time.Now(),
 			})
@@ -142,9 +160,10 @@ func (factory *tcpStreamFactory) getStreamProps(srcIP string, srcPort string, ds
 	}
 }
 
+//lint:ignore U1000 will be used in the future
 func (factory *tcpStreamFactory) shouldNotifyOnOutboundLink(dstIP string, dstPort int) bool {
 	if inArrayInt(remoteOnlyOutboundPorts, dstPort) {
-		isDirectedHere := inArrayString(ownIps, dstIP)
+		isDirectedHere := inArrayString(factory.ownIps, dstIP)
 		return !isDirectedHere && !isPrivateIP(dstIP)
 	}
 	return true

tap/tcp_streams_map.go

@@ -0,0 +1,71 @@
+package tap
+
+import (
+	"runtime"
+	_debug "runtime/debug"
+	"sync"
+	"time"
+
+	"github.com/up9inc/mizu/shared/logger"
+)
+
+type tcpStreamMap struct {
+	streams  *sync.Map
+	streamId int64
+}
+
+func NewTcpStreamMap() *tcpStreamMap {
+	return &tcpStreamMap{
+		streams: &sync.Map{},
+	}
+}
+
+func (streamMap *tcpStreamMap) Store(key, value interface{}) {
+	streamMap.streams.Store(key, value)
+}
+
+func (streamMap *tcpStreamMap) Delete(key interface{}) {
+	streamMap.streams.Delete(key)
+}
+
+func (streamMap *tcpStreamMap) nextId() int64 {
+	streamMap.streamId++
+	return streamMap.streamId
+}
+
+func (streamMap *tcpStreamMap) closeTimedoutTcpStreamChannels() {
+	tcpStreamChannelTimeout := GetTcpChannelTimeoutMs()
+	for {
+		time.Sleep(10 * time.Millisecond)
+		_debug.FreeOSMemory()
+		streamMap.streams.Range(func(key interface{}, value interface{}) bool {
+			streamWrapper := value.(*tcpStreamWrapper)
+			stream := streamWrapper.stream
+			if stream.superIdentifier.Protocol == nil {
+				if !stream.isClosed && time.Now().After(streamWrapper.createdAt.Add(tcpStreamChannelTimeout)) {
+					stream.Close()
+					appStats.IncDroppedTcpStreams()
+					logger.Log.Debugf("Dropped an unidentified TCP stream because of timeout. Total dropped: %d Total Goroutines: %d Timeout (ms): %d\n",
+						appStats.DroppedTcpStreams, runtime.NumGoroutine(), tcpStreamChannelTimeout/1000000)
+				}
+			} else {
+				if !stream.superIdentifier.IsClosedOthers {
+					for i := range stream.clients {
+						reader := &stream.clients[i]
+						if reader.extension.Protocol != stream.superIdentifier.Protocol {
+							reader.Close()
+						}
+					}
+					for i := range stream.servers {
+						reader := &stream.servers[i]
+						if reader.extension.Protocol != stream.superIdentifier.Protocol {
+							reader.Close()
+						}
+					}
+					stream.superIdentifier.IsClosedOthers = true
+				}
+			}
+			return true
+		})
+	}
+}

tap/tester/.gitignore

@@ -0,0 +1 @@
+tester

tap/tester/README.md

@@ -0,0 +1,12 @@
+
+This tester used to launch passive-tapper locally without Docker or Kuberenetes environment.
+
+Its good for testing purposes.
+
+# How to run
+
+From the `tap` folder run:
+`./tester/launch.sh`
+
+The tester gets the same arguments the passive_tapper gets, run with `--help` to get a complete list of options.
+`./tester/launch.sh --help`

tap/tester/launch.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+set -e
+
+echo "Building extensions..."
+pushd .. && ./devops/build_extensions.sh && popd
+
+go build -o tester tester/tester.go
+
+sudo ./tester/tester "$@"

tap/tester/tester.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path"
+	"plugin"
+	"sort"
+	"strings"
+
+	"github.com/op/go-logging"
+
+	"github.com/go-errors/errors"
+	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap"
+	tapApi "github.com/up9inc/mizu/tap/api"
+)
+
+func loadExtensions() ([]*tapApi.Extension, error) {
+	extensionsDir := "./extensions"
+	files, err := ioutil.ReadDir(extensionsDir)
+
+	if err != nil {
+		return nil, errors.Wrap(err, 0)
+	}
+
+	extensions := make([]*tapApi.Extension, 0)
+	for _, file := range files {
+		filename := file.Name()
+
+		if !strings.HasSuffix(filename, ".so") {
+			continue
+		}
+
+		fmt.Printf("Loading extension: %s\n", filename)
+
+		extension := &tapApi.Extension{
+			Path: path.Join(extensionsDir, filename),
+		}
+
+		plug, err := plugin.Open(extension.Path)
+
+		if err != nil {
+			return nil, errors.Wrap(err, 0)
+		}
+
+		extension.Plug = plug
+		symDissector, err := plug.Lookup("Dissector")
+
+		if err != nil {
+			return nil, errors.Wrap(err, 0)
+		}
+
+		dissector, ok := symDissector.(tapApi.Dissector)
+
+		if !ok {
+			return nil, errors.Errorf("Symbol Dissector type error: %v %T\n", file, symDissector)
+		}
+
+		dissector.Register(extension)
+		extension.Dissector = dissector
+		extensions = append(extensions, extension)
+	}
+
+	sort.Slice(extensions, func(i, j int) bool {
+		return extensions[i].Protocol.Priority < extensions[j].Protocol.Priority
+	})
+
+	for _, extension := range extensions {
+		fmt.Printf("Extension Properties: %+v\n", extension)
+	}
+
+	return extensions, nil
+}
+
+func internalRun() error {
+	logger.InitLoggerStderrOnly(logging.DEBUG)
+
+	opts := tap.TapOpts{
+		HostMode: false,
+	}
+
+	outputItems := make(chan *tapApi.OutputChannelItem, 1000)
+	extenssions, err := loadExtensions()
+
+	if err != nil {
+		return err
+	}
+
+	tapOpts := tapApi.TrafficFilteringOptions{}
+
+	tap.StartPassiveTapper(&opts, outputItems, extenssions, &tapOpts)
+
+	fmt.Printf("Tapping, press enter to exit...\n")
+	reader := bufio.NewReader(os.Stdin)
+	reader.ReadLine()
+	return nil
+}
+
+func main() {
+	err := internalRun()
+
+	if err != nil {
+		switch err := err.(type) {
+		case *errors.Error:
+			fmt.Printf("Error: %v\n", err.ErrorStack())
+		default:
+			fmt.Printf("Error: %v\n", err)
+		}
+
+		os.Exit(1)
+	}
+}

ui/src/components/EntryDetailed.tsx

@@ -71,7 +71,7 @@ export const EntryDetailed: React.FC<EntryDetailedProps> = ({entryData}) => {
         />
         {entryData.data && <EntrySummary data={entryData.data}/>}
         <>
-            {entryData.data && <EntryViewer representation={entryData.representation} isRulesEnabled={entryData.isRulesEnabled} rulesMatched={entryData.rulesMatched} elapsedTime={entryData.data.elapsedTime} color={entryData.protocol.backgroundColor}/>}
+            {entryData.data && <EntryViewer representation={entryData.representation} isRulesEnabled={entryData.isRulesEnabled} rulesMatched={entryData.rulesMatched} contractStatus={entryData.data.contractStatus} requestReason={entryData.data.contractRequestReason} responseReason={entryData.data.contractResponseReason} contractContent={entryData.data.contractContent} elapsedTime={entryData.data.elapsedTime} color={entryData.protocol.backgroundColor}/>}
         </>
     </>
 };

ui/src/components/EntryDetailed/EntrySections.tsx

@@ -150,8 +150,6 @@ export const EntryTableSection: React.FC<EntrySectionProps> = ({title, color, ar
     </React.Fragment>
 }
 
-
-
 interface EntryPolicySectionProps {
     title: string,
     color: string,
@@ -159,7 +157,6 @@ interface EntryPolicySectionProps {
     arrayToIterate: any[],
 }
 
-
 interface EntryPolicySectionCollapsibleTitleProps {
     label: string;
     matched: string;
@@ -208,7 +205,7 @@ export const EntryTablePolicySection: React.FC<EntryPolicySectionProps> = ({titl
                         <tbody>
                             {arrayToIterate.map(({rule, matched}, index) => {
                                     return (
-                                        <EntryPolicySectionContainer key={index} label={rule.Name} matched={matched && (rule.Type === 'latency' ? rule.Latency >= latency : true)? "Success" : "Failure"}>
+                                        <EntryPolicySectionContainer key={index} label={rule.Name} matched={matched && (rule.Type === 'slo' ? rule.ResponseTime >= latency : true)? "Success" : "Failure"}>
                                             {
                                                 <>
                                                     {
@@ -216,8 +213,8 @@ export const EntryTablePolicySection: React.FC<EntryPolicySectionProps> = ({titl
                                                         <tr className={styles.dataValue}><td><b>Key:</b></td> <td>{rule.Key}</td></tr>
                                                     }
                                                     {
-                                                        rule.Latency !== 0 &&
-                                                        <tr className={styles.dataValue}><td><b>Latency:</b></td> <td>{rule.Latency}</td></tr>
+                                                        rule.ResponseTime !== 0 &&
+                                                        <tr className={styles.dataValue}><td><b>Response Time:</b></td> <td>{rule.ResponseTime}</td></tr>
                                                     }
                                                     {
                                                         rule.Method &&
@@ -253,3 +250,28 @@ export const EntryTablePolicySection: React.FC<EntryPolicySectionProps> = ({titl
         }
     </React.Fragment>
 }
+
+interface EntryContractSectionProps {
+    color: string,
+    requestReason: string,
+    responseReason: string,
+    contractContent: string,
+}
+
+export const EntryContractSection: React.FC<EntryContractSectionProps> = ({color, requestReason, responseReason, contractContent}) => {
+    return <React.Fragment>
+        {requestReason && <EntrySectionContainer title="Request" color={color}>
+            {requestReason}
+        </EntrySectionContainer>}
+        {responseReason && <EntrySectionContainer title="Response" color={color}>
+            {responseReason}
+        </EntrySectionContainer>}
+        {contractContent && <EntrySectionContainer title="Contract" color={color}>
+            <SyntaxHighlighter
+                isWrapped={false}
+                code={contractContent}
+                language={"yaml"}
+            />
+        </EntrySectionContainer>}
+    </React.Fragment>
+}

ui/src/components/EntryDetailed/EntryViewer.tsx

@@ -1,7 +1,7 @@
 import React, {useState} from 'react';
 import styles from './EntryViewer.module.sass';
 import Tabs from "../UI/Tabs";
-import {EntryTableSection, EntryBodySection, EntryTablePolicySection} from "./EntrySections";
+import {EntryTableSection, EntryBodySection, EntryTablePolicySection, EntryContractSection} from "./EntrySections";
 
 enum SectionTypes {
     SectionTable = "table",
@@ -33,7 +33,7 @@ const SectionsRepresentation: React.FC<any> = ({data, color}) => {
     return <>{sections}</>;
 }
 
-const AutoRepresentation: React.FC<any> = ({representation, isRulesEnabled, rulesMatched, elapsedTime, color}) => {
+const AutoRepresentation: React.FC<any> = ({representation, isRulesEnabled, rulesMatched, contractStatus, requestReason, responseReason, contractContent, elapsedTime, color}) => {
     var TABS = [
         {
             tab: 'Request'
@@ -50,6 +50,7 @@ const AutoRepresentation: React.FC<any> = ({representation, isRulesEnabled, rule
 
     var responseTabIndex = 0;
     var rulesTabIndex = 0;
+    var contractTabIndex = 0;
 
     if (response) {
         TABS.push(
@@ -69,11 +70,19 @@ const AutoRepresentation: React.FC<any> = ({representation, isRulesEnabled, rule
         rulesTabIndex = TABS.length - 1;
     }
 
+    if (contractStatus !== 0 && contractContent) {
+        TABS.push(
+            {
+                tab: 'Contract',
+            }
+        );
+        contractTabIndex = TABS.length - 1;
+    }
+
     return <div className={styles.Entry}>
         {<div className={styles.body}>
             <div className={styles.bodyHeader}>
                 <Tabs tabs={TABS} currentTab={currentTab} color={color} onChange={setCurrentTab} leftAligned/>
-                {request?.url && <a className={styles.endpointURL} href={request.payload.url} target='_blank' rel="noreferrer">{request.payload.url}</a>}
             </div>
             {currentTab === TABS[0].tab && <React.Fragment>
                 <SectionsRepresentation data={request} color={color}/>
@@ -84,6 +93,9 @@ const AutoRepresentation: React.FC<any> = ({representation, isRulesEnabled, rule
             {isRulesEnabled && currentTab === TABS[rulesTabIndex].tab && <React.Fragment>
                 <EntryTablePolicySection title={'Rule'} color={color} latency={elapsedTime} arrayToIterate={rulesMatched ? rulesMatched : []}/>
             </React.Fragment>}
+            {contractStatus !== 0 && contractContent && currentTab === TABS[contractTabIndex].tab && <React.Fragment>
+                <EntryContractSection color={color} requestReason={requestReason} responseReason={responseReason} contractContent={contractContent}/>
+            </React.Fragment>}
         </div>}
     </div>;
 }
@@ -92,12 +104,26 @@ interface Props {
     representation: any;
     isRulesEnabled: boolean;
     rulesMatched: any;
+    contractStatus: number;
+    requestReason: string;
+    responseReason: string;
+    contractContent: string;
     color: string;
     elapsedTime: number;
 }
 
-const EntryViewer: React.FC<Props> = ({representation, isRulesEnabled, rulesMatched, elapsedTime, color}) => {
-    return <AutoRepresentation representation={representation} isRulesEnabled={isRulesEnabled} rulesMatched={rulesMatched} elapsedTime={elapsedTime} color={color}/>
+const EntryViewer: React.FC<Props> = ({representation, isRulesEnabled, rulesMatched, contractStatus, requestReason, responseReason, contractContent, elapsedTime, color}) => {
+    return <AutoRepresentation
+        representation={representation}
+        isRulesEnabled={isRulesEnabled}
+        rulesMatched={rulesMatched}
+        contractStatus={contractStatus}
+        requestReason={requestReason}
+        responseReason={responseReason}
+        contractContent={contractContent}
+        elapsedTime={elapsedTime}
+        color={color}
+    />
 };
 
 export default EntryViewer;

ui/src/components/EntryListItem/EntryListItem.module.sass

@@ -38,6 +38,7 @@
 .ruleNumberText
   font-size: 12px
   font-weight: 600
+  white-space: nowrap
 
 .ruleNumberTextFailure
   color: #DB2156
@@ -72,12 +73,17 @@
   padding-left: 10px
   flex-grow: 1
 
-.directionContainer
+.separatorRight
   display: flex
   border-right: 1px solid $data-background-color
   padding: 4px
   padding-right: 12px
 
+.separatorLeft
+  display: flex
+  padding: 4px
+  padding-left: 12px
+
 .port
   font-size: 12px
   color: $secondary-font-color

ui/src/components/EntryListItem/EntryListItem.tsx

@@ -26,6 +26,7 @@ interface Entry {
     isOutgoing?: boolean;
     latency: number;
     rules: Rules;
+    contractStatus: number,
 }
 
 interface Rules {
@@ -64,7 +65,7 @@ export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, isSel
         }
     }
     let additionalRulesProperties = "";
-    let ruleSuccess: boolean;
+    let ruleSuccess = true;
     let rule = 'latency' in entry.rules
     if (rule) {
         if (entry.rules.latency !== -1) {
@@ -91,11 +92,32 @@ export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, isSel
             }
         }
     }
+
+    var contractEnabled = true;
+    var contractText = "";
+    switch (entry.contractStatus) {
+        case 0:
+            contractEnabled = false;
+            break;
+        case 1:
+            additionalRulesProperties = styles.ruleSuccessRow
+            ruleSuccess = true
+            contractText = "No Breaches"
+            break;
+        case 2:
+            additionalRulesProperties = styles.ruleFailureRow
+            ruleSuccess = false
+            contractText = "Breach"
+            break;
+        default:
+            break;
+    }
+
     return <>
         <div
             id={entry.id}
             className={`${styles.row}
-            ${isSelected && !rule ? styles.rowSelected : additionalRulesProperties}`}
+            ${isSelected && !rule && !contractEnabled ? styles.rowSelected : additionalRulesProperties}`}
             onClick={() => setFocusedEntryId(entry.id)}
             style={{
                 border: isSelected ? `1px ${entry.protocol.backgroundColor} solid` : "1px transparent solid",
@@ -117,12 +139,19 @@ export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, isSel
             </div>
             {
                 rule ?
-                    <div className={`${styles.ruleNumberText} ${ruleSuccess ? styles.ruleNumberTextSuccess : styles.ruleNumberTextFailure}`}>
+                    <div className={`${styles.ruleNumberText} ${ruleSuccess ? styles.ruleNumberTextSuccess : styles.ruleNumberTextFailure} ${rule && contractEnabled ? styles.separatorRight : ""}`}>
                         {`Rules (${numberOfRules})`}
                     </div>
                 : ""
             }
-            <div className={styles.directionContainer}>
+            {
+                contractEnabled ?
+                    <div className={`${styles.ruleNumberText} ${ruleSuccess ? styles.ruleNumberTextSuccess : styles.ruleNumberTextFailure} ${rule && contractEnabled ? styles.separatorLeft : ""}`}>
+                        {contractText}
+                    </div>
+                : ""
+            }
+            <div className={styles.separatorRight}>
                 <span className={styles.port} title="Source Port">{entry.sourcePort}</span>
                 {entry.isOutgoing ?
                     <img src={outgoingIcon} alt="Ingoing traffic" title="Ingoing"/>