remove duplicated tcp packet source (#405 )

added alert and error when there is a proxy before k8b server (#402 )
Tapper Refactor (#396 )
2026-02-19 20:40:17 +00:00 · 2021-10-26 18:16:33 +03:00 · 2021-10-26 16:57:56 +03:00 · 2021-10-25 13:59:06 +03:00 · 2021-10-24 15:10:46 +03:00 · 2021-10-24 12:27:05 +03:00
31 changed files with 933 additions and 488 deletions
--- a/.github/workflows/publish.yml
+++ b/.github/workflows/publish.yml
@@ -14,6 +14,10 @@ jobs:
  docker:
    runs-on: ubuntu-latest
    steps:
+      - name: Set up Go 1.16
+        uses: actions/setup-go@v2
+        with:
+          go-version: '1.16'
      - name: Checkout
        uses: actions/checkout@v2
      - name: Set up Cloud SDK
--- a/README.md
+++ b/README.md
@@ -15,6 +15,10 @@ Think TCPDump and Chrome Dev Tools combined.
 - No installation or code instrumentation
 - Works completely on premises

+## Requirements
+
+A Kubernetes server version of 1.16.0 or higher is required.
+
 ## Download

 Download Mizu for your platform and operating system
--- a/agent/go.sum
+++ b/agent/go.sum
@@ -79,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=
--- a/agent/pkg/resolver/resolver.go
+++ b/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 {
--- a/agent/pkg/rules/rulesHTTP.go
+++ b/agent/pkg/rules/rulesHTTP.go
@@ -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
 				}
--- a/agent/pkg/up9/main.go
+++ b/agent/pkg/up9/main.go
@@ -3,6 +3,7 @@ package up9
 import (
 	"bytes"
 	"compress/zlib"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -231,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)
 			}

--- a/cli/Makefile
+++ b/cli/Makefile
@@ -27,8 +27,9 @@ build-all: ## Build for all supported platforms.
 	@mkdir -p bin && sed s/_SEM_VER_/$(SEM_VER)/g README.md.TEMPLATE >  bin/README.md
 	@$(MAKE) build GOOS=darwin GOARCH=amd64
 	@$(MAKE) build GOOS=linux GOARCH=amd64
-	@# $(MAKE) build GOOS=darwin GOARCH=arm64
-	@# $(MAKE) GOOS=windows GOARCH=amd64
+	@$(MAKE) build GOOS=darwin GOARCH=arm64
+	@$(MAKE) build GOOS=windows GOARCH=amd64
+	@mv ./bin/mizu_windows_amd64 ./bin/mizu.exe
 	@# $(MAKE) GOOS=linux GOARCH=386
 	@# $(MAKE) GOOS=windows GOARCH=386
 	@# $(MAKE) GOOS=linux GOARCH=arm64
--- a/cli/README.md.TEMPLATE
+++ b/cli/README.md.TEMPLATE
@@ -2,16 +2,25 @@

 Download Mizu for your platform

-**Mac** (on Intel chip)
+**Mac** (Intel)
 ```
 curl -Lo mizu https://github.com/up9inc/mizu/releases/download/_SEM_VER_/mizu_darwin_amd64 && chmod 755 mizu
 ```

+**Mac** (Apple M1 silicon)
+```
+curl -Lo mizu https://github.com/up9inc/mizu/releases/download/_SEM_VER_/mizu_darwin_arm64 && chmod 755 mizu
+```
+
 **Linux** 
 ```
 curl -Lo mizu https://github.com/up9inc/mizu/releases/download/_SEM_VER_/mizu_linux_amd64 && chmod 755 mizu
 ```

+**Windows** (Intel 64bit)
+```
+curl -LO https://github.com/up9inc/mizu/releases/download/_SEM_VER_/mizu.exe
+```

 ### Checksums
 SHA256 checksums available for compiled binaries.
--- a/cli/config/configStruct.go
+++ b/cli/config/configStruct.go
@@ -14,6 +14,7 @@ import (
 const (
 	MizuResourcesNamespaceConfigName = "mizu-resources-namespace"
 	ConfigFilePathCommandName        = "config-path"
+	KubeConfigPathConfigName         = "kube-config-path"
 )

 type ConfigStruct struct {
--- a/cli/kubernetes/provider.go
+++ b/cli/kubernetes/provider.go
@@ -7,6 +7,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"k8s.io/apimachinery/pkg/version"
+	"net/url"
 	"path/filepath"
 	"regexp"
 	"strconv"
@@ -77,6 +79,10 @@ func NewProvider(kubeConfigPath string) (*Provider, error) {
 			"you can set alternative kube config file path by adding the kube-config-path field to the mizu config file, err:  %w", kubeConfigPath, err)
 	}

+	if err := validateNotProxy(kubernetesConfig, restClientConfig); err != nil {
+		return nil, err
+	}
+
 	return &Provider{
 		clientSet:        clientSet,
 		kubernetesConfig: kubernetesConfig,
@@ -727,3 +733,31 @@ func loadKubernetesConfiguration(kubeConfigPath string) clientcmd.ClientConfig {
 func isPodRunning(pod *core.Pod) bool {
 	return pod.Status.Phase == core.PodRunning
 }
+
+// We added this after a customer tried to run mizu from lens, which used len's kube config, which have cluster server configuration, which points to len's local proxy.
+// The workaround was to use the user's local default kube config.
+// For now - we are blocking the option to run mizu through a proxy to k8s server
+func validateNotProxy(kubernetesConfig clientcmd.ClientConfig, restClientConfig *restclient.Config) error {
+	kubernetesUrl, err := url.Parse(restClientConfig.Host)
+	if err != nil {
+		logger.Log.Debugf("validateNotProxy - error while parsing kubernetes host, err: %v", err)
+		return nil
+	}
+
+	restProxyClientConfig, _ := kubernetesConfig.ClientConfig()
+	restProxyClientConfig.Host = kubernetesUrl.Host
+
+	clientProxySet, err := getClientSet(restProxyClientConfig)
+	if err == nil {
+		proxyServerVersion, err := clientProxySet.ServerVersion()
+		if err != nil {
+			return nil
+		}
+
+		if *proxyServerVersion == (version.Info{}) {
+			return fmt.Errorf("cannot establish http-proxy connection to the Kubernetes cluster. If you’re using Lens or similar tool, please run mizu with regular kubectl config using --%v %v=$HOME/.kube/config flag", config.SetCommandName, config.KubeConfigPathConfigName)
+		}
+	}
+
+	return nil
+}
--- a/cli/mizu/version/versionCheck.go
+++ b/cli/mizu/version/versionCheck.go
@@ -85,7 +85,13 @@ func CheckNewerVersion(versionChan chan string) {
 	logger.Log.Debugf("Finished version validation, github version %v, current version %v, took %v", gitHubVersion, currentSemVer, time.Since(start))

 	if gitHubVersionSemVer.GreaterThan(currentSemVer) {
-		versionChan <- fmt.Sprintf("Update available! %v -> %v (curl -Lo mizu %v/mizu_%s_amd64 && chmod 755 mizu)", mizu.SemVer, gitHubVersion, strings.Replace(*latestRelease.HTMLURL, "tag", "download", 1), runtime.GOOS)
+		var downloadMessage string
+		if runtime.GOOS == "windows" {
+			downloadMessage = fmt.Sprintf("curl -LO %v/mizu.exe", strings.Replace(*latestRelease.HTMLURL, "tag", "download", 1))
+		} else {
+			downloadMessage = fmt.Sprintf("curl -Lo mizu %v/mizu_%s_%s && chmod 755 mizu", strings.Replace(*latestRelease.HTMLURL, "tag", "download", 1), runtime.GOOS, runtime.GOARCH)
+		}
+		versionChan <- fmt.Sprintf("Update available! %v -> %v (%s)", mizu.SemVer, gitHubVersion, downloadMessage)
 	} else {
 		versionChan <- ""
 	}
--- a/tap/cleaner.go
+++ b/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()
 		}
--- a/tap/diagnose/diagnose.go
+++ b/tap/diagnose/diagnose.go
@@ -0,0 +1,76 @@
+package diagnose
+
+import (
+	"fmt"
+	"os"
+	"runtime"
+	"runtime/pprof"
+	"strconv"
+	"time"
+
+	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/api"
+)
+
+var AppStats = api.AppStats{}
+
+func StartMemoryProfiler(envDumpPath string, envTimeInterval string) {
+	dumpPath := "/app/pprof"
+	if envDumpPath != "" {
+		dumpPath = envDumpPath
+	}
+	timeInterval := 60
+	if envTimeInterval != "" {
+		if i, err := strconv.Atoi(envTimeInterval); err == nil {
+			timeInterval = i
+		}
+	}
+
+	logger.Log.Info("Profiling is on, results will be written to %s", dumpPath)
+	go func() {
+		if _, err := os.Stat(dumpPath); os.IsNotExist(err) {
+			if err := os.Mkdir(dumpPath, 0777); err != nil {
+				logger.Log.Fatal("could not create directory for profile: ", err)
+			}
+		}
+
+		for {
+			t := time.Now()
+
+			filename := fmt.Sprintf("%s/%s__mem.prof", dumpPath, t.Format("15_04_05"))
+
+			logger.Log.Infof("Writing memory profile to %s\n", filename)
+
+			f, err := os.Create(filename)
+			if err != nil {
+				logger.Log.Fatal("could not create memory profile: ", err)
+			}
+			runtime.GC() // get up-to-date statistics
+			if err := pprof.WriteHeapProfile(f); err != nil {
+				logger.Log.Fatal("could not write memory profile: ", err)
+			}
+			_ = f.Close()
+			time.Sleep(time.Second * time.Duration(timeInterval))
+		}
+	}()
+}
+
+func DumpMemoryProfile(filename string) error {
+	if filename == "" {
+		return nil
+	}
+
+	f, err := os.Create(filename)
+
+	if err != nil {
+		return err
+	}
+
+	defer f.Close()
+
+	if err := pprof.WriteHeapProfile(f); err != nil {
+		return err
+	}
+
+	return nil
+}
--- a/tap/diagnose/errors_map.go
+++ b/tap/diagnose/errors_map.go
@@ -0,0 +1,85 @@
+package diagnose
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/google/gopacket/examples/util"
+	"github.com/up9inc/mizu/shared/logger"
+)
+
+var TapErrors *errorsMap
+
+type errorsMap struct {
+	errorsMap      map[string]uint
+	OutputLevel    int
+	ErrorsCount    uint
+	errorsMapMutex sync.Mutex
+}
+
+func InitializeErrorsMap(debug bool, verbose bool, quiet bool) {
+	var outputLevel int
+
+	defer util.Run()()
+	if debug {
+		outputLevel = 2
+	} else if verbose {
+		outputLevel = 1
+	} else if quiet {
+		outputLevel = -1
+	}
+
+	TapErrors = newErrorsMap(outputLevel)
+}
+
+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.ErrorsCount++
+	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
+}
+
+func (e *errorsMap) PrintSummary() {
+	logger.Log.Infof("Errors: %d", e.ErrorsCount)
+	for t := range e.errorsMap {
+		logger.Log.Infof(" %s:\t\t%d", e, e.errorsMap[t])
+	}
+}
--- a/tap/diagnose/internal_stats.go
+++ b/tap/diagnose/internal_stats.go
@@ -0,0 +1,46 @@
+package diagnose
+
+import "github.com/up9inc/mizu/shared/logger"
+
+type tapperInternalStats struct {
+	Ipdefrag            int
+	MissedBytes         int
+	Pkt                 int
+	Sz                  int
+	Totalsz             int
+	RejectFsm           int
+	RejectOpt           int
+	RejectConnFsm       int
+	Reassembled         int
+	OutOfOrderBytes     int
+	OutOfOrderPackets   int
+	BiggestChunkBytes   int
+	BiggestChunkPackets int
+	OverlapBytes        int
+	OverlapPackets      int
+}
+
+var InternalStats *tapperInternalStats
+
+func InitializeTapperInternalStats() {
+	InternalStats = &tapperInternalStats{}
+}
+
+func (stats *tapperInternalStats) PrintStatsSummary() {
+	logger.Log.Infof("IPdefrag:\t\t%d", stats.Ipdefrag)
+	logger.Log.Infof("TCP stats:")
+	logger.Log.Infof(" missed bytes:\t\t%d", stats.MissedBytes)
+	logger.Log.Infof(" total packets:\t\t%d", stats.Pkt)
+	logger.Log.Infof(" rejected FSM:\t\t%d", stats.RejectFsm)
+	logger.Log.Infof(" rejected Options:\t%d", stats.RejectOpt)
+	logger.Log.Infof(" reassembled bytes:\t%d", stats.Sz)
+	logger.Log.Infof(" total TCP bytes:\t%d", stats.Totalsz)
+	logger.Log.Infof(" conn rejected FSM:\t%d", stats.RejectConnFsm)
+	logger.Log.Infof(" reassembled chunks:\t%d", stats.Reassembled)
+	logger.Log.Infof(" out-of-order packets:\t%d", stats.OutOfOrderPackets)
+	logger.Log.Infof(" out-of-order bytes:\t%d", stats.OutOfOrderBytes)
+	logger.Log.Infof(" biggest-chunk packets:\t%d", stats.BiggestChunkPackets)
+	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)
+}
--- a/tap/go.mod
+++ b/tap/go.mod
@@ -4,7 +4,9 @@ go 1.16

 require (
 	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/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 // indirect
--- a/tap/go.sum
+++ b/tap/go.sum
@@ -1,6 +1,8 @@
 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=
--- a/tap/net_utils.go
+++ b/tap/net_utils.go
@@ -3,6 +3,8 @@ package tap
 import (
 	"net"
 	"strings"
+
+	"github.com/up9inc/mizu/tap/diagnose"
 )

 var privateIPBlocks []*net.IPNet
@@ -27,6 +29,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 +57,7 @@ func initPrivateIPBlocks() {
 	} {
 		_, block, err := net.ParseCIDR(cidr)
 		if err != nil {
-			Error("Private-IP-Block-Parse", "parse error on %q: %v", cidr, err)
+			diagnose.TapErrors.Error("Private-IP-Block-Parse", "parse error on %q: %v", cidr, err)
 		} else {
 			privateIPBlocks = append(privateIPBlocks, block)
 		}
--- a/tap/outboundlinks.go
+++ b/tap/outboundlinks.go
@@ -7,11 +7,11 @@ const (
 )

 type OutboundLink struct {
-	Src     string
-	DstIP   string
-	DstPort int
+	Src                   string
+	DstIP                 string
+	DstPort               int
 	SuggestedResolvedName string
-	SuggestedProtocol OutboundLinkProtocol
+	SuggestedProtocol     OutboundLinkProtocol
 }

 func NewOutboundLinkWriter() *OutboundLinkWriter {
@@ -26,11 +26,11 @@ type OutboundLinkWriter struct {

 func (olw *OutboundLinkWriter) WriteOutboundLink(src string, DstIP string, DstPort int, SuggestedResolvedName string, SuggestedProtocol OutboundLinkProtocol) {
 	olw.OutChan <- &OutboundLink{
-		Src: src,
-		DstIP: DstIP,
-		DstPort: DstPort,
+		Src:                   src,
+		DstIP:                 DstIP,
+		DstPort:               DstPort,
 		SuggestedResolvedName: SuggestedResolvedName,
-		SuggestedProtocol: SuggestedProtocol,
+		SuggestedProtocol:     SuggestedProtocol,
 	}
 }

--- a/tap/passive_tapper.go
+++ b/tap/passive_tapper.go
@@ -9,33 +9,22 @@
 package tap

 import (
-	"encoding/hex"
 	"encoding/json"
 	"flag"
-	"fmt"
-	"io"
 	"os"
-	"os/signal"
 	"runtime"
-	_debug "runtime/debug"
-	"runtime/pprof"
-	"strconv"
 	"strings"
-	"sync"
 	"time"

-	"github.com/google/gopacket"
-	"github.com/google/gopacket/examples/util"
-	"github.com/google/gopacket/ip4defrag"
-	"github.com/google/gopacket/layers" // pulls in all layers decoders
-	"github.com/google/gopacket/pcap"
-	"github.com/google/gopacket/reassembly"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
+	"github.com/up9inc/mizu/tap/diagnose"
+	"github.com/up9inc/mizu/tap/source"
 )

 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")
@@ -62,69 +51,14 @@ var staleTimeoutSeconds = flag.Int("staletimout", 120, "Max time in seconds to k

 var memprofile = flag.String("memprofile", "", "Write memory profile")

-var appStats = api.AppStats{}
-
-// global
-var stats struct {
-	ipdefrag            int
-	missedBytes         int
-	pkt                 int
-	sz                  int
-	totalsz             int
-	rejectFsm           int
-	rejectOpt           int
-	rejectConnFsm       int
-	reassembled         int
-	outOfOrderBytes     int
-	outOfOrderPackets   int
-	biggestChunkBytes   int
-	biggestChunkPackets int
-	overlapBytes        int
-	overlapPackets      int
-}
-
 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 inArrayInt(arr []int, valueToCheck int) bool {
 	for _, value := range arr {
 		if value == valueToCheck {
@@ -143,397 +77,121 @@ func inArrayString(arr []string, valueToCheck string) bool {
 	return false
 }

-// Context
-// The assembler context
-type Context struct {
-	CaptureInfo gopacket.CaptureInfo
-}
-
-func GetStats() api.AppStats {
-	return appStats
-}
-
-func (c *Context) GetCaptureInfo() gopacket.CaptureInfo {
-	return c.CaptureInfo
-}
-
 func StartPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem, extensionsRef []*api.Extension, options *api.TrafficFilteringOptions) {
 	hostMode = opts.HostMode
 	extensions = extensionsRef
 	filteringOptions = options

 	if GetMemoryProfilingEnabled() {
-		startMemoryProfiler()
+		diagnose.StartMemoryProfiler(os.Getenv(MemoryProfilingDumpPath), os.Getenv(MemoryProfilingTimeIntervalSeconds))
 	}

 	go startPassiveTapper(outputItems)
 }

-func startMemoryProfiler() {
-	dumpPath := "/app/pprof"
-	envDumpPath := os.Getenv(MemoryProfilingDumpPath)
-	if envDumpPath != "" {
-		dumpPath = envDumpPath
-	}
-	timeInterval := 60
-	envTimeInterval := os.Getenv(MemoryProfilingTimeIntervalSeconds)
-	if envTimeInterval != "" {
-		if i, err := strconv.Atoi(envTimeInterval); err == nil {
-			timeInterval = i
-		}
-	}
+func printPeriodicStats(cleaner *Cleaner) {
+	statsPeriod := time.Second * time.Duration(*statsevery)
+	ticker := time.NewTicker(statsPeriod)

-	logger.Log.Info("Profiling is on, results will be written to %s", dumpPath)
-	go func() {
-		if _, err := os.Stat(dumpPath); os.IsNotExist(err) {
-			if err := os.Mkdir(dumpPath, 0777); err != nil {
-				logger.Log.Fatal("could not create directory for profile: ", err)
-			}
-		}
-
-		for true {
-			t := time.Now()
-
-			filename := fmt.Sprintf("%s/%s__mem.prof", dumpPath, t.Format("15_04_05"))
-
-			logger.Log.Infof("Writing memory profile to %s\n", filename)
-
-			f, err := os.Create(filename)
-			if err != nil {
-				logger.Log.Fatal("could not create memory profile: ", err)
-			}
-			runtime.GC() // get up-to-date statistics
-			if err := pprof.WriteHeapProfile(f); err != nil {
-				logger.Log.Fatal("could not write memory profile: ", err)
-			}
-			_ = f.Close()
-			time.Sleep(time.Second * time.Duration(timeInterval))
-		}
-	}()
-}
-
-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
-		})
+		<-ticker.C
+
+		// Since the start
+		errorMapLen, errorsSummery := diagnose.TapErrors.GetErrorsSummary()
+
+		logger.Log.Infof("%v (errors: %v, errTypes:%v) - Errors Summary: %s",
+			time.Since(diagnose.AppStats.StartTime),
+			diagnose.TapErrors.ErrorsCount,
+			errorMapLen,
+			errorsSummery,
+		)
+
+		// At this moment
+		memStats := runtime.MemStats{}
+		runtime.ReadMemStats(&memStats)
+		logger.Log.Infof(
+			"mem: %d, goroutines: %d",
+			memStats.HeapAlloc,
+			runtime.NumGoroutine(),
+		)
+
+		// Since the last print
+		cleanStats := cleaner.dumpStats()
+		logger.Log.Infof(
+			"cleaner - flushed connections: %d, closed connections: %d, deleted messages: %d",
+			cleanStats.flushed,
+			cleanStats.closed,
+			cleanStats.deleted,
+		)
+		currentAppStats := diagnose.AppStats.DumpStats()
+		appStatsJSON, _ := json.Marshal(currentAppStats)
+		logger.Log.Infof("app stats - %v", string(appStatsJSON))
 	}
 }

 func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
-	go closeTimedoutTcpStreamChannels()
+	streamsMap := NewTcpStreamMap()
+	go streamsMap.closeTimedoutTcpStreamChannels()

-	defer util.Run()()
-	if *debug {
-		outputLevel = 2
-	} else if *verbose {
-		outputLevel = 1
-	} else if *quiet {
-		outputLevel = -1
-	}
-	errorsMap = make(map[string]uint)
+	diagnose.InitializeErrorsMap(*debug, *verbose, *quiet)
+	diagnose.InitializeTapperInternalStats()

-	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
-	}
-
-	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)
-		}
-	} else {
-		// This is a little complicated because we want to allow all possible options
-		// for creating the packet capture handle... instead of all this you can
-		// just call pcap.OpenLive if you want a simple handle.
-		inactive, err := pcap.NewInactiveHandle(*iface)
-		if err != nil {
-			logger.Log.Fatalf("could not create: %v", err)
-		}
-		defer inactive.CleanUp()
-		if err = inactive.SetSnapLen(*snaplen); err != nil {
-			logger.Log.Fatalf("could not set snap length: %v", err)
-		} else if err = inactive.SetPromisc(*promisc); err != nil {
-			logger.Log.Fatalf("could not set promisc mode: %v", err)
-		} else if err = inactive.SetTimeout(time.Second); err != nil {
-			logger.Log.Fatalf("could not set timeout: %v", err)
-		}
-		if *tstype != "" {
-			if t, err := pcap.TimestampSourceFromString(*tstype); err != nil {
-				logger.Log.Fatalf("Supported timestamp types: %v", inactive.SupportedTimestamps())
-			} else if err := inactive.SetTimestampSource(t); err != nil {
-				logger.Log.Fatalf("Supported timestamp types: %v", inactive.SupportedTimestamps())
-			}
-		}
-		if handle, err = inactive.Activate(); err != nil {
-			logger.Log.Fatalf("PCAP Activate error: %v", err)
-		}
-		defer handle.Close()
-	}
+	var bpffilter string
 	if len(flag.Args()) > 0 {
-		bpffilter := strings.Join(flag.Args(), " ")
-		logger.Log.Infof("Using BPF filter %q", bpffilter)
-		if err = handle.SetBPFFilter(bpffilter); err != nil {
-			logger.Log.Fatalf("BPF filter error: %v", err)
-		}
+		bpffilter = strings.Join(flag.Args(), " ")
 	}

-	var dec gopacket.Decoder
-	var ok bool
-	decoderName := *decoder
-	if decoderName == "" {
-		decoderName = fmt.Sprintf("%s", handle.LinkType())
+	packetSource, err := source.NewTcpPacketSource(*fname, *iface, source.TcpPacketSourceBehaviour{
+		SnapLength:  *snaplen,
+		Promisc:     *promisc,
+		Tstype:      *tstype,
+		DecoderName: *decoder,
+		Lazy:        *lazy,
+		BpfFilter:   bpffilter,
+	})
+
+	if err != nil {
+		logger.Log.Fatal(err)
 	}
-	if dec, ok = gopacket.DecodersByLayerName[decoderName]; !ok {
-		logger.Log.Fatal("No decoder named", decoderName)
+
+	defer packetSource.Close()
+
+	if err != nil {
+		logger.Log.Fatal(err)
 	}
-	source := gopacket.NewPacketSource(handle, dec)
-	source.Lazy = *lazy
-	source.NoCopy = true
+
+	packets := make(chan source.TcpPacketInfo)
+	assembler := NewTcpAssembler(outputItems, streamsMap)
+
 	logger.Log.Info("Starting to read packets")
-	appStats.SetStartTime(time.Now())
-	defragger := ip4defrag.NewIPv4Defragmenter()
+	diagnose.AppStats.SetStartTime(time.Now())

-	var emitter api.Emitter = &api.Emitting{
-		AppStats:      &appStats,
-		OutputChannel: outputItems,
-	}
-
-	streamFactory := &tcpStreamFactory{
-		Emitter: emitter,
-	}
-	streamPool := reassembly.NewStreamPool(streamFactory)
-	assembler := reassembly.NewAssembler(streamPool)
-
-	maxBufferedPagesTotal := GetMaxBufferedPagesPerConnection()
-	maxBufferedPagesPerConnection := GetMaxBufferedPagesTotal()
-	logger.Log.Infof("Assembler options: maxBufferedPagesTotal=%d, maxBufferedPagesPerConnection=%d", maxBufferedPagesTotal, maxBufferedPagesPerConnection)
-	assembler.AssemblerOptions.MaxBufferedPagesTotal = maxBufferedPagesTotal
-	assembler.AssemblerOptions.MaxBufferedPagesPerConnection = maxBufferedPagesPerConnection
-
-	var assemblerMutex sync.Mutex
-
-	signalChan := make(chan os.Signal, 1)
-	signal.Notify(signalChan, os.Interrupt)
+	go packetSource.ReadPackets(!*nodefrag, packets)

 	staleConnectionTimeout := time.Second * time.Duration(*staleTimeoutSeconds)
 	cleaner := Cleaner{
-		assembler:         assembler,
-		assemblerMutex:    &assemblerMutex,
+		assembler:         assembler.Assembler,
+		assemblerMutex:    &assembler.assemblerMutex,
 		cleanPeriod:       cleanPeriod,
 		connectionTimeout: staleConnectionTimeout,
 	}
 	cleaner.start()

-	go func() {
-		statsPeriod := time.Second * time.Duration(*statsevery)
-		ticker := time.NewTicker(statsPeriod)
+	go printPeriodicStats(&cleaner)

-		for true {
-			<-ticker.C
+	assembler.processPackets(*hexdumppkt, packets)

-			// Since the start
-			errorsMapMutex.Lock()
-			errorMapLen := len(errorsMap)
-			errorsSummery := fmt.Sprintf("%v", errorsMap)
-			errorsMapMutex.Unlock()
-			logger.Log.Infof("%v (errors: %v, errTypes:%v) - Errors Summary: %s",
-				time.Since(appStats.StartTime),
-				nErrors,
-				errorMapLen,
-				errorsSummery,
-			)
-
-			// At this moment
-			memStats := runtime.MemStats{}
-			runtime.ReadMemStats(&memStats)
-			logger.Log.Infof(
-				"mem: %d, goroutines: %d",
-				memStats.HeapAlloc,
-				runtime.NumGoroutine(),
-			)
-
-			// Since the last print
-			cleanStats := cleaner.dumpStats()
-			logger.Log.Infof(
-				"cleaner - flushed connections: %d, closed connections: %d, deleted messages: %d",
-				cleanStats.flushed,
-				cleanStats.closed,
-				cleanStats.deleted,
-			)
-			currentAppStats := appStats.DumpStats()
-			appStatsJSON, _ := json.Marshal(currentAppStats)
-			logger.Log.Infof("app stats - %v", string(appStatsJSON))
-		}
-	}()
-
-	if GetMemoryProfilingEnabled() {
-		startMemoryProfiler()
+	if diagnose.TapErrors.OutputLevel >= 2 {
+		assembler.dumpStreamPool()
 	}

-	for {
-		packet, err := source.NextPacket()
-		if err == io.EOF {
-			break
-		} else if err != nil {
-			logger.Log.Debugf("Error: %v", err)
-			continue
-		}
-		packetsCount := appStats.IncPacketsCount()
-		logger.Log.Debugf("PACKET #%d", packetsCount)
-		data := packet.Data()
-		appStats.UpdateProcessedBytes(uint64(len(data)))
-		if *hexdumppkt {
-			logger.Log.Debugf("Packet content (%d/0x%x) - %s", len(data), len(data), hex.Dump(data))
-		}
-
-		// defrag the IPv4 packet if required
-		if !*nodefrag {
-			ip4Layer := packet.Layer(layers.LayerTypeIPv4)
-			if ip4Layer == nil {
-				continue
-			}
-			ip4 := ip4Layer.(*layers.IPv4)
-			l := ip4.Length
-			newip4, err := defragger.DefragIPv4(ip4)
-			if err != nil {
-				logger.Log.Fatal("Error while de-fragmenting", err)
-			} else if newip4 == nil {
-				logger.Log.Debugf("Fragment...")
-				continue // packet fragment, we don't have whole packet yet.
-			}
-			if newip4.Length != l {
-				stats.ipdefrag++
-				logger.Log.Debugf("Decoding re-assembled packet: %s", newip4.NextLayerType())
-				pb, ok := packet.(gopacket.PacketBuilder)
-				if !ok {
-					logger.Log.Panic("Not a PacketBuilder")
-				}
-				nextDecoder := newip4.NextLayerType()
-				_ = nextDecoder.Decode(newip4.Payload, pb)
-			}
-		}
-
-		tcp := packet.Layer(layers.LayerTypeTCP)
-		if tcp != nil {
-			appStats.IncTcpPacketsCount()
-			tcp := tcp.(*layers.TCP)
-			if *checksum {
-				err := tcp.SetNetworkLayerForChecksum(packet.NetworkLayer())
-				if err != nil {
-					logger.Log.Fatalf("Failed to set network layer for checksum: %s\n", err)
-				}
-			}
-			c := Context{
-				CaptureInfo: packet.Metadata().CaptureInfo,
-			}
-			stats.totalsz += len(tcp.Payload)
-			logger.Log.Debugf("%s : %v -> %s : %v", packet.NetworkLayer().NetworkFlow().Src(), tcp.SrcPort, packet.NetworkLayer().NetworkFlow().Dst(), tcp.DstPort)
-			assemblerMutex.Lock()
-			assembler.AssembleWithContext(packet.NetworkLayer().NetworkFlow(), tcp, &c)
-			assemblerMutex.Unlock()
-		}
-
-		done := *maxcount > 0 && int64(appStats.PacketsCount) >= *maxcount
-		if done {
-			errorsMapMutex.Lock()
-			errorMapLen := len(errorsMap)
-			errorsMapMutex.Unlock()
-			logger.Log.Infof("Processed %v packets (%v bytes) in %v (errors: %v, errTypes:%v)",
-				appStats.PacketsCount,
-				appStats.ProcessedBytes,
-				time.Since(appStats.StartTime),
-				nErrors,
-				errorMapLen)
-		}
-		select {
-		case <-signalChan:
-			logger.Log.Infof("Caught SIGINT: aborting")
-			done = true
-		default:
-			// NOP: continue
-		}
-		if done {
-			break
-		}
+	if err := diagnose.DumpMemoryProfile(*memprofile); err != nil {
+		logger.Log.Errorf("Error dumping memory profile %v\n", err)
 	}

-	assemblerMutex.Lock()
-	closed := assembler.FlushAll()
-	assemblerMutex.Unlock()
-	logger.Log.Debugf("Final flush: %d closed", closed)
-	if outputLevel >= 2 {
-		streamPool.Dump()
-	}
+	assembler.waitAndDump()

-	if *memprofile != "" {
-		f, err := os.Create(*memprofile)
-		if err != nil {
-			logger.Log.Fatal(err)
-		}
-		_ = pprof.WriteHeapProfile(f)
-		_ = f.Close()
-	}
-
-	streamFactory.WaitGoRoutines()
-	assemblerMutex.Lock()
-	logger.Log.Debugf("%s", assembler.Dump())
-	assemblerMutex.Unlock()
-	if !*nodefrag {
-		logger.Log.Infof("IPdefrag:\t\t%d", stats.ipdefrag)
-	}
-	logger.Log.Infof("TCP stats:")
-	logger.Log.Infof(" missed bytes:\t\t%d", stats.missedBytes)
-	logger.Log.Infof(" total packets:\t\t%d", stats.pkt)
-	logger.Log.Infof(" rejected FSM:\t\t%d", stats.rejectFsm)
-	logger.Log.Infof(" rejected Options:\t%d", stats.rejectOpt)
-	logger.Log.Infof(" reassembled bytes:\t%d", stats.sz)
-	logger.Log.Infof(" total TCP bytes:\t%d", stats.totalsz)
-	logger.Log.Infof(" conn rejected FSM:\t%d", stats.rejectConnFsm)
-	logger.Log.Infof(" reassembled chunks:\t%d", stats.reassembled)
-	logger.Log.Infof(" out-of-order packets:\t%d", stats.outOfOrderPackets)
-	logger.Log.Infof(" out-of-order bytes:\t%d", stats.outOfOrderBytes)
-	logger.Log.Infof(" biggest-chunk packets:\t%d", stats.biggestChunkPackets)
-	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("AppStats: %v", GetStats())
+	diagnose.InternalStats.PrintStatsSummary()
+	diagnose.TapErrors.PrintSummary()
+	logger.Log.Infof("AppStats: %v", diagnose.AppStats)
 }
--- a/tap/source/tcp_packet_source.go
+++ b/tap/source/tcp_packet_source.go
@@ -0,0 +1,150 @@
+package source
+
+import (
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/ip4defrag"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcap"
+	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/diagnose"
+)
+
+type TcpPacketSource struct {
+	source    *gopacket.PacketSource
+	handle    *pcap.Handle
+	defragger *ip4defrag.IPv4Defragmenter
+	Behaviour *TcpPacketSourceBehaviour
+}
+
+type TcpPacketSourceBehaviour struct {
+	SnapLength  int
+	Promisc     bool
+	Tstype      string
+	DecoderName string
+	Lazy        bool
+	BpfFilter   string
+}
+
+type TcpPacketInfo struct {
+	Packet gopacket.Packet
+	Source *TcpPacketSource
+}
+
+func NewTcpPacketSource(filename string, interfaceName string,
+	behaviour TcpPacketSourceBehaviour) (*TcpPacketSource, error) {
+	var err error
+
+	result := &TcpPacketSource{
+		defragger: ip4defrag.NewIPv4Defragmenter(),
+		Behaviour: &behaviour,
+	}
+
+	if filename != "" {
+		if result.handle, err = pcap.OpenOffline(filename); err != nil {
+			return result, fmt.Errorf("PCAP OpenOffline error: %v", err)
+		}
+	} else {
+		// This is a little complicated because we want to allow all possible options
+		// for creating the packet capture handle... instead of all this you can
+		// just call pcap.OpenLive if you want a simple handle.
+		inactive, err := pcap.NewInactiveHandle(interfaceName)
+		if err != nil {
+			return result, fmt.Errorf("could not create: %v", err)
+		}
+		defer inactive.CleanUp()
+		if err = inactive.SetSnapLen(behaviour.SnapLength); err != nil {
+			return result, fmt.Errorf("could not set snap length: %v", err)
+		} else if err = inactive.SetPromisc(behaviour.Promisc); err != nil {
+			return result, fmt.Errorf("could not set promisc mode: %v", err)
+		} else if err = inactive.SetTimeout(time.Second); err != nil {
+			return result, fmt.Errorf("could not set timeout: %v", err)
+		}
+		if behaviour.Tstype != "" {
+			if t, err := pcap.TimestampSourceFromString(behaviour.Tstype); err != nil {
+				return result, fmt.Errorf("supported timestamp types: %v", inactive.SupportedTimestamps())
+			} else if err := inactive.SetTimestampSource(t); err != nil {
+				return result, fmt.Errorf("supported timestamp types: %v", inactive.SupportedTimestamps())
+			}
+		}
+		if result.handle, err = inactive.Activate(); err != nil {
+			return result, fmt.Errorf("PCAP Activate error: %v", err)
+		}
+	}
+	if behaviour.BpfFilter != "" {
+		logger.Log.Infof("Using BPF filter %q", behaviour.BpfFilter)
+		if err = result.handle.SetBPFFilter(behaviour.BpfFilter); err != nil {
+			return nil, fmt.Errorf("BPF filter error: %v", err)
+		}
+	}
+
+	var dec gopacket.Decoder
+	var ok bool
+	if behaviour.DecoderName == "" {
+		behaviour.DecoderName = result.handle.LinkType().String()
+	}
+	if dec, ok = gopacket.DecodersByLayerName[behaviour.DecoderName]; !ok {
+		return nil, fmt.Errorf("no decoder named %v", behaviour.DecoderName)
+	}
+	result.source = gopacket.NewPacketSource(result.handle, dec)
+	result.source.Lazy = behaviour.Lazy
+	result.source.NoCopy = true
+
+	return result, nil
+}
+
+func (source *TcpPacketSource) Close() {
+	if source.handle != nil {
+		source.handle.Close()
+	}
+}
+
+func (source *TcpPacketSource) ReadPackets(ipdefrag bool, packets chan<- TcpPacketInfo) error {
+	for {
+		packet, err := source.source.NextPacket()
+
+		if err == io.EOF {
+			return err
+		} else if err != nil {
+			if err.Error() != "Timeout Expired" {
+				logger.Log.Debugf("Error: %T", err)
+			}
+			continue
+		}
+
+		// defrag the IPv4 packet if required
+		if !ipdefrag {
+			ip4Layer := packet.Layer(layers.LayerTypeIPv4)
+			if ip4Layer == nil {
+				continue
+			}
+			ip4 := ip4Layer.(*layers.IPv4)
+			l := ip4.Length
+			newip4, err := source.defragger.DefragIPv4(ip4)
+			if err != nil {
+				logger.Log.Fatal("Error while de-fragmenting", err)
+			} else if newip4 == nil {
+				logger.Log.Debugf("Fragment...")
+				continue // packet fragment, we don't have whole packet yet.
+			}
+			if newip4.Length != l {
+				diagnose.InternalStats.Ipdefrag++
+				logger.Log.Debugf("Decoding re-assembled packet: %s", newip4.NextLayerType())
+				pb, ok := packet.(gopacket.PacketBuilder)
+				if !ok {
+					logger.Log.Panic("Not a PacketBuilder")
+				}
+				nextDecoder := newip4.NextLayerType()
+				_ = nextDecoder.Decode(newip4.Payload, pb)
+			}
+		}
+
+		packets <- TcpPacketInfo{
+			Packet: packet,
+			Source: source,
+		}
+	}
+}
--- a/tap/tcp_assembler.go
+++ b/tap/tcp_assembler.go
@@ -0,0 +1,132 @@
+package tap
+
+import (
+	"encoding/hex"
+	"os"
+	"os/signal"
+	"sync"
+	"time"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/reassembly"
+	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/api"
+	"github.com/up9inc/mizu/tap/diagnose"
+	"github.com/up9inc/mizu/tap/source"
+)
+
+type tcpAssembler struct {
+	*reassembly.Assembler
+	streamPool     *reassembly.StreamPool
+	streamFactory  *tcpStreamFactory
+	assemblerMutex sync.Mutex
+}
+
+// Context
+// The assembler context
+type context struct {
+	CaptureInfo gopacket.CaptureInfo
+}
+
+func (c *context) GetCaptureInfo() gopacket.CaptureInfo {
+	return c.CaptureInfo
+}
+
+func NewTcpAssembler(outputItems chan *api.OutputChannelItem, streamsMap *tcpStreamMap) *tcpAssembler {
+	var emitter api.Emitter = &api.Emitting{
+		AppStats:      &diagnose.AppStats,
+		OutputChannel: outputItems,
+	}
+
+	streamFactory := NewTcpStreamFactory(emitter, streamsMap)
+	streamPool := reassembly.NewStreamPool(streamFactory)
+	assembler := reassembly.NewAssembler(streamPool)
+
+	maxBufferedPagesTotal := GetMaxBufferedPagesPerConnection()
+	maxBufferedPagesPerConnection := GetMaxBufferedPagesTotal()
+	logger.Log.Infof("Assembler options: maxBufferedPagesTotal=%d, maxBufferedPagesPerConnection=%d",
+		maxBufferedPagesTotal, maxBufferedPagesPerConnection)
+	assembler.AssemblerOptions.MaxBufferedPagesTotal = maxBufferedPagesTotal
+	assembler.AssemblerOptions.MaxBufferedPagesPerConnection = maxBufferedPagesPerConnection
+
+	return &tcpAssembler{
+		Assembler:     assembler,
+		streamPool:    streamPool,
+		streamFactory: streamFactory,
+	}
+}
+
+func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.TcpPacketInfo) {
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, os.Interrupt)
+
+	for packetInfo := range packets {
+		packetsCount := diagnose.AppStats.IncPacketsCount()
+		logger.Log.Debugf("PACKET #%d", packetsCount)
+		packet := packetInfo.Packet
+		data := packet.Data()
+		diagnose.AppStats.UpdateProcessedBytes(uint64(len(data)))
+		if dumpPacket {
+			logger.Log.Debugf("Packet content (%d/0x%x) - %s", len(data), len(data), hex.Dump(data))
+		}
+
+		tcp := packet.Layer(layers.LayerTypeTCP)
+		if tcp != nil {
+			diagnose.AppStats.IncTcpPacketsCount()
+			tcp := tcp.(*layers.TCP)
+			if *checksum {
+				err := tcp.SetNetworkLayerForChecksum(packet.NetworkLayer())
+				if err != nil {
+					logger.Log.Fatalf("Failed to set network layer for checksum: %s\n", err)
+				}
+			}
+			c := context{
+				CaptureInfo: packet.Metadata().CaptureInfo,
+			}
+			diagnose.InternalStats.Totalsz += len(tcp.Payload)
+			logger.Log.Debugf("%s : %v -> %s : %v", packet.NetworkLayer().NetworkFlow().Src(), tcp.SrcPort, packet.NetworkLayer().NetworkFlow().Dst(), tcp.DstPort)
+			a.assemblerMutex.Lock()
+			a.AssembleWithContext(packet.NetworkLayer().NetworkFlow(), tcp, &c)
+			a.assemblerMutex.Unlock()
+		}
+
+		done := *maxcount > 0 && int64(diagnose.AppStats.PacketsCount) >= *maxcount
+		if done {
+			errorMapLen, _ := diagnose.TapErrors.GetErrorsSummary()
+			logger.Log.Infof("Processed %v packets (%v bytes) in %v (errors: %v, errTypes:%v)",
+				diagnose.AppStats.PacketsCount,
+				diagnose.AppStats.ProcessedBytes,
+				time.Since(diagnose.AppStats.StartTime),
+				diagnose.TapErrors.ErrorsCount,
+				errorMapLen)
+		}
+
+		select {
+		case <-signalChan:
+			logger.Log.Infof("Caught SIGINT: aborting")
+			done = true
+		default:
+			// NOP: continue
+		}
+		if done {
+			break
+		}
+	}
+
+	a.assemblerMutex.Lock()
+	closed := a.FlushAll()
+	a.assemblerMutex.Unlock()
+	logger.Log.Debugf("Final flush: %d closed", closed)
+}
+
+func (a *tcpAssembler) dumpStreamPool() {
+	a.streamPool.Dump()
+}
+
+func (a *tcpAssembler) waitAndDump() {
+	a.streamFactory.WaitGoRoutines()
+	a.assemblerMutex.Lock()
+	logger.Log.Debugf("%s", a.Dump())
+	a.assemblerMutex.Unlock()
+}
--- a/tap/tcp_reader.go
+++ b/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
--- a/tap/tcp_stream.go
+++ b/tap/tcp_stream.go
@@ -9,6 +9,7 @@ import (
 	"github.com/google/gopacket/layers" // pulls in all layers decoders
 	"github.com/google/gopacket/reassembly"
 	"github.com/up9inc/mizu/tap/api"
+	"github.com/up9inc/mizu/tap/diagnose"
 )

 /* It's a connection (bidirectional)
@@ -28,19 +29,19 @@ 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())
-		stats.rejectFsm++
+		diagnose.TapErrors.SilentError("FSM-rejection", "%s: Packet rejected by FSM (state:%s)", t.ident, t.tcpstate.String())
+		diagnose.InternalStats.RejectFsm++
 		if !t.fsmerr {
 			t.fsmerr = true
-			stats.rejectConnFsm++
+			diagnose.InternalStats.RejectConnFsm++
 		}
 		if !*ignorefsmerr {
 			return false
@@ -49,8 +50,8 @@ 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)
-		stats.rejectOpt++
+		diagnose.TapErrors.SilentError("OptionChecker-rejection", "%s: Packet rejected by OptionChecker: %s", t.ident, err)
+		diagnose.InternalStats.RejectOpt++
 		if !*nooptcheck {
 			return false
 		}
@@ -60,15 +61,15 @@ 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)
+			diagnose.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)
+			diagnose.TapErrors.SilentError("Checksum", "%s: Invalid checksum: 0x%x", t.ident, c)
 			accept = false
 		}
 	}
 	if !accept {
-		stats.rejectOpt++
+		diagnose.InternalStats.RejectOpt++
 	}
 	return accept
 }
@@ -79,28 +80,28 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 	// update stats
 	sgStats := sg.Stats()
 	if skip > 0 {
-		stats.missedBytes += skip
+		diagnose.InternalStats.MissedBytes += skip
 	}
-	stats.sz += length - saved
-	stats.pkt += sgStats.Packets
+	diagnose.InternalStats.Sz += length - saved
+	diagnose.InternalStats.Pkt += sgStats.Packets
 	if sgStats.Chunks > 1 {
-		stats.reassembled++
+		diagnose.InternalStats.Reassembled++
 	}
-	stats.outOfOrderPackets += sgStats.QueuedPackets
-	stats.outOfOrderBytes += sgStats.QueuedBytes
-	if length > stats.biggestChunkBytes {
-		stats.biggestChunkBytes = length
+	diagnose.InternalStats.OutOfOrderPackets += sgStats.QueuedPackets
+	diagnose.InternalStats.OutOfOrderBytes += sgStats.QueuedBytes
+	if length > diagnose.InternalStats.BiggestChunkBytes {
+		diagnose.InternalStats.BiggestChunkBytes = length
 	}
-	if sgStats.Packets > stats.biggestChunkPackets {
-		stats.biggestChunkPackets = sgStats.Packets
+	if sgStats.Packets > diagnose.InternalStats.BiggestChunkPackets {
+		diagnose.InternalStats.BiggestChunkPackets = sgStats.Packets
 	}
 	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)
+		diagnose.TapErrors.SilentError("Invalid-Overlap", "bytes:%d, pkts:%d", sgStats.OverlapBytes, sgStats.OverlapPackets)
 	}
-	stats.overlapBytes += sgStats.OverlapBytes
-	stats.overlapPackets += sgStats.OverlapPackets
+	diagnose.InternalStats.OverlapBytes += sgStats.OverlapBytes
+	diagnose.InternalStats.OverlapPackets += sgStats.OverlapPackets

 	var ident string
 	if dir == reassembly.TCPDirClientToServer {
@@ -108,7 +109,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)
 	}
-	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)
+	diagnose.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 +128,18 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		}
 		dnsSize := binary.BigEndian.Uint16(data[:2])
 		missing := int(dnsSize) - len(data[2:])
-		Debug("dnsSize: %d, missing: %d", dnsSize, missing)
+		diagnose.TapErrors.Debug("dnsSize: %d, missing: %d", dnsSize, missing)
 		if missing > 0 {
-			Debug("Missing some bytes: %d", missing)
+			diagnose.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)
+			diagnose.TapErrors.SilentError("DNS-parser", "Failed to decode DNS: %v", err)
 		} else {
-			Debug("DNS: %s", gopacket.LayerDump(dns))
+			diagnose.TapErrors.Debug("DNS: %s", gopacket.LayerDump(dns))
 		}
 		if len(data) > 2+int(dnsSize) {
 			sg.KeepFrom(2 + int(dnsSize))
@@ -147,7 +148,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		if length > 0 {
 			// This is where we pass the reassembled information onwards
 			// This channel is read by an tcpReader object
-			appStats.IncReassembledTcpPayloadsCount()
+			diagnose.AppStats.IncReassembledTcpPayloadsCount()
 			timestamp := ac.GetCaptureInfo().Timestamp
 			if dir == reassembly.TCPDirClientToServer {
 				for i := range t.clients {
@@ -173,7 +174,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 }

 func (t *tcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
-	Debug("%s: Connection closed", t.ident)
+	diagnose.TapErrors.Debug("%s: Connection closed", t.ident)
 	if t.isTapTarget && !t.isClosed {
 		t.Close()
 	}
@@ -193,7 +194,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]
--- a/tap/tcp_stream_factory.go
+++ b/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
--- a/tap/tcp_streams_map.go
+++ b/tap/tcp_streams_map.go
@@ -0,0 +1,72 @@
+package tap
+
+import (
+	"runtime"
+	_debug "runtime/debug"
+	"sync"
+	"time"
+
+	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/diagnose"
+)
+
+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()
+					diagnose.AppStats.IncDroppedTcpStreams()
+					logger.Log.Debugf("Dropped an unidentified TCP stream because of timeout. Total dropped: %d Total Goroutines: %d Timeout (ms): %d\n",
+						diagnose.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
+		})
+	}
+}
--- a/tap/tester/.gitignore
+++ b/tap/tester/.gitignore
@@ -0,0 +1 @@
+tester
--- a/tap/tester/README.md
+++ b/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`
--- a/tap/tester/launch.sh
+++ b/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 "$@"
--- a/tap/tester/tester.go
+++ b/tap/tester/tester.go
@@ -0,0 +1,114 @@
+package main
+
+import (
+	"bufio"
+	"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
+		}
+
+		logger.Log.Infof("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 {
+		logger.Log.Infof("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)
+
+	logger.Log.Infof("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:
+			logger.Log.Errorf("Error: %v\n", err.ErrorStack())
+		default:
+			logger.Log.Errorf("Error: %v\n", err)
+		}
+
+		os.Exit(1)
+	}
+}
--- a/ui/src/components/EntryDetailed/EntrySections.tsx
+++ b/ui/src/components/EntryDetailed/EntrySections.tsx
@@ -205,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"}>
                                            {
                                                <>
                                                    {
@@ -213,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 &&
Author	SHA1	Message	Date
David Levanon	e635b97d11	remove duplicated tcp packet source (#405 )	2021-10-26 18:16:33 +03:00
RoyUP9	779257b864	added alert and error when there is a proxy before k8b server (#402 )	2021-10-26 16:57:56 +03:00
David Levanon	e696851261	Tapper Refactor (#396 ) * introduce tcp_assembler and tcp_packet_source - the motivation is to … (#380) * add passive-tapper main tester (#353) * add passive-tapper main tester * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * rename main to tester * build extenssions as part of the tester launch * add a README to the tester * solving go.mod and .sum conflicts with addition of go-errors * trivial warning fixes (#354) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * tcp streams map (#355) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * change rlog to mizu logger * errors map (#356) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * change int to uint - errorsmap * change from int to uint * Change errorsMap.nErrors to uint. * change errors map to mizu logger instead of rlog * init mizu logger in tester + fix errormap declaration Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * move own ips to tcp stream factory (#358) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * Feature/tapper refactor i/move own ips to tcp stream factory (#379) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * fix ownips compilation issue * introduce tcp_assembler and tcp_packet_source - the motivation is to split the actual packet sources from the assembler, so we can have a single thread for the assembly which is separated from packet source threads * make struts private at this point - planning to move some packages to their own package so we can utilize encapsulation * move context to tcp_assembly + fix error check of tcp source packet * use param instead of gloab flag for ipdefrag * fix a typo * remove unused pid param Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * extract stats functions out of the main tapping function (#381) * add passive-tapper main tester (#353) * add passive-tapper main tester * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * rename main to tester * build extenssions as part of the tester launch * add a README to the tester * solving go.mod and .sum conflicts with addition of go-errors * trivial warning fixes (#354) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * tcp streams map (#355) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * change rlog to mizu logger * errors map (#356) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * change int to uint - errorsmap * change from int to uint * Change errorsMap.nErrors to uint. * change errors map to mizu logger instead of rlog * init mizu logger in tester + fix errormap declaration Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * move own ips to tcp stream factory (#358) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * Feature/tapper refactor i/move own ips to tcp stream factory (#379) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * fix ownips compilation issue * introduce tcp_assembler and tcp_packet_source - the motivation is to split the actual packet sources from the assembler, so we can have a single thread for the assembly which is separated from packet source threads * make struts private at this point - planning to move some packages to their own package so we can utilize encapsulation * extract stats functions out of the main tapping function * move context to tcp_assembly + fix error check of tcp source packet * use param instead of gloab flag for ipdefrag Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * Feature/tapper refactor i/internal tapper stats (#384) * add passive-tapper main tester (#353) * add passive-tapper main tester * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * rename main to tester * build extenssions as part of the tester launch * add a README to the tester * solving go.mod and .sum conflicts with addition of go-errors * trivial warning fixes (#354) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * tcp streams map (#355) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * change rlog to mizu logger * errors map (#356) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * change int to uint - errorsmap * change from int to uint * Change errorsMap.nErrors to uint. * change errors map to mizu logger instead of rlog * init mizu logger in tester + fix errormap declaration Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * move own ips to tcp stream factory (#358) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * Feature/tapper refactor i/move own ips to tcp stream factory (#379) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * fix ownips compilation issue * introduce tcp_assembler and tcp_packet_source - the motivation is to split the actual packet sources from the assembler, so we can have a single thread for the assembly which is separated from packet source threads * make struts private at this point - planning to move some packages to their own package so we can utilize encapsulation * extract stats functions out of the main tapping function * move context to tcp_assembly + fix error check of tcp source packet * use param instead of gloab flag for ipdefrag * introduce internal tapper stats * minor fixes for errors map and internal stats Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * Feature/tapper refactor i/diagnose package (#386) * add passive-tapper main tester (#353) * add passive-tapper main tester * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * rename main to tester * build extenssions as part of the tester launch * add a README to the tester * solving go.mod and .sum conflicts with addition of go-errors * trivial warning fixes (#354) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * tcp streams map (#355) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * change rlog to mizu logger * errors map (#356) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * change int to uint - errorsmap * change from int to uint * Change errorsMap.nErrors to uint. * change errors map to mizu logger instead of rlog * init mizu logger in tester + fix errormap declaration Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * move own ips to tcp stream factory (#358) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * Feature/tapper refactor i/move own ips to tcp stream factory (#379) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * fix ownips compilation issue * introduce tcp_assembler and tcp_packet_source - the motivation is to split the actual packet sources from the assembler, so we can have a single thread for the assembly which is separated from packet source threads * make struts private at this point - planning to move some packages to their own package so we can utilize encapsulation * extract stats functions out of the main tapping function * move context to tcp_assembly + fix error check of tcp source packet * use param instead of gloab flag for ipdefrag * introduce internal tapper stats * minor fixes for errors map and internal stats * move errors map + app stats + internal stats + periodic tasks to diagnose package * initialize tapper internal stats Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * move tcp packet source to its packet (#387) * add passive-tapper main tester (#353) * add passive-tapper main tester * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * rename main to tester * build extenssions as part of the tester launch * add a README to the tester * solving go.mod and .sum conflicts with addition of go-errors * trivial warning fixes (#354) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * tcp streams map (#355) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * change rlog to mizu logger * errors map (#356) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * change int to uint - errorsmap * change from int to uint * Change errorsMap.nErrors to uint. * change errors map to mizu logger instead of rlog * init mizu logger in tester + fix errormap declaration Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * move own ips to tcp stream factory (#358) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * Feature/tapper refactor i/move own ips to tcp stream factory (#379) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * fix ownips compilation issue * introduce tcp_assembler and tcp_packet_source - the motivation is to split the actual packet sources from the assembler, so we can have a single thread for the assembly which is separated from packet source threads * make struts private at this point - planning to move some packages to their own package so we can utilize encapsulation * extract stats functions out of the main tapping function * move context to tcp_assembly + fix error check of tcp source packet * use param instead of gloab flag for ipdefrag * introduce internal tapper stats * minor fixes for errors map and internal stats * move errors map + app stats + internal stats + periodic tasks to diagnose package * move tcp packet source to its packet * initialize tapper internal stats Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * Fix coding style * Remove `tap/internal_stats.go` * make channel between input and assembler blocking - to preserve the same behaviour we have before the refactor Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> Co-authored-by: M. Mert Yildiran <mehmet@up9.com>	2021-10-25 13:59:06 +03:00
RoyUP9	4e50e17d81	build m1 and windows (#395 )	2021-10-24 15:10:46 +03:00
Igor Gov	991eb2ab16	Revert "TRA-3828 added build for Mac/Apple M1 and Windows (#392 )" (#393 ) This reverts commit `f0db3b81a8`.	2021-10-24 12:27:05 +03:00
Alex Haiut	f0db3b81a8	TRA-3828 added build for Mac/Apple M1 and Windows (#392 )	2021-10-24 11:56:12 +03:00
Nimrod Gilboa Markevich	9df1812d8e	Add k8s version requirements to README (#389 ) A version lower than 1.16.0 fails with the error message: Error updating tappers: 415: Unsupported Media Type.	2021-10-21 15:29:26 +03:00
RoyUP9	4f6da91d74	fixed naming of latency to response time (#388 )	2021-10-21 12:45:17 +03:00
David Levanon	e2e69a3dc4	remove main dir (#385 )	2021-10-20 13:48:53 +03:00
RoyUP9	b6db64d868	fixed sync entries text (#383 )	2021-10-20 12:25:31 +03:00
RamiBerm	160ae77145	TRA-3811 fix service resolving (#382 ) Co-authored-by: Rami Berman <rami.berman@up9.com>	2021-10-20 11:50:39 +03:00
David Levanon	2944493e2d	passive-tapper refactor - first phase * add passive-tapper main tester (#353) * add passive-tapper main tester * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * rename main to tester * build extenssions as part of the tester launch * add a README to the tester * solving go.mod and .sum conflicts with addition of go-errors * trivial warning fixes (#354) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * disable host mode for tester - to avoid filterAuthorities * tcp streams map (#355) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * change rlog to mizu logger * errors map (#356) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * change int to uint - errorsmap * change from int to uint * Change errorsMap.nErrors to uint. * change errors map to mizu logger instead of rlog * init mizu logger in tester + fix errormap declaration Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com> * move own ips to tcp stream factory (#358) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * Feature/tapper refactor i/move own ips to tcp stream factory (#379) * add passive-tapper main tester * trivial warning fixes * add errors to go.sum of mizu agent * tcp streams map * disable host mode for tester - to avoid filterAuthorities * set tcp streams map for tcp stream factory * errors map * move own ips to tcp stream factory * fix ownips compilation issue Co-authored-by: Nimrod Gilboa Markevich <nimrod@up9.com>	2021-10-20 11:15:22 +03:00