upgrade alpine base image (#413)

Develop -> Main #patch

Dockerfile

@@ -44,7 +44,7 @@ RUN go build -ldflags="-s -w \
 COPY devops/build_extensions.sh ..
 RUN cd .. && /bin/bash build_extensions.sh
 
-FROM alpine:3.13.5
+FROM alpine:3.14
 
 RUN apk add bash libpcap-dev tcpdump
 WORKDIR /app

Makefile

@@ -46,10 +46,6 @@ push-docker: ## Build and publish agent docker image.
 	@echo "publishing Docker image .. "
 	devops/build-push-featurebranch.sh
 
-push-docker-debug:
-	@echo "publishing debug Docker image .. "
-	devops/build-push-featurebranch-debug.sh
-
 build-docker-ci: ## Build agent docker image for CI.
 	@echo "building docker image for ci"
 	devops/build-agent-ci.sh

agent/pkg/database/main.go

@@ -13,12 +13,11 @@ import (
 )
 
 const (
-	DBPath     = "./entries.db"
-	OrderDesc  = "desc"
-	OrderAsc   = "asc"
-	LT         = "lt"
-	GT         = "gt"
-	TimeFormat = "2006-01-02 15:04:05.000000000"
+	DBPath    = "./entries.db"
+	OrderDesc = "desc"
+	OrderAsc  = "asc"
+	LT        = "lt"
+	GT        = "gt"
 )
 
 var (
@@ -58,7 +57,7 @@ func initDataBase(databasePath string) *gorm.DB {
 	return temp
 }
 
-func GetEntriesFromDb(timeFrom time.Time, timeTo time.Time, protocolName *string) []tapApi.MizuEntry {
+func GetEntriesFromDb(timestampFrom int64, timestampTo int64, protocolName *string) []tapApi.MizuEntry {
 	order := OrderDesc
 	protocolNameCondition := "1 = 1"
 	if protocolName != nil {
@@ -68,7 +67,7 @@ func GetEntriesFromDb(timeFrom time.Time, timeTo time.Time, protocolName *string
 	var entries []tapApi.MizuEntry
 	GetEntriesTable().
 		Where(protocolNameCondition).
-		Where(fmt.Sprintf("created_at BETWEEN '%s' AND '%s'", timeFrom.Format(TimeFormat), timeTo.Format(TimeFormat))).
+		Where(fmt.Sprintf("timestamp BETWEEN %v AND %v", timestampFrom, timestampTo)).
 		Order(fmt.Sprintf("timestamp %s", order)).
 		Find(&entries)
 

agent/pkg/up9/main.go

@@ -206,13 +206,13 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 
 	sleepTime := time.Second * time.Duration(uploadIntervalSec)
 
-	var timeFrom time.Time
-	protocolFilter := "http"
+	var timestampFrom int64 = 0
 
 	for {
-		timeTo := time.Now()
-		logger.Log.Infof("Getting entries from %v, to %v\n", timeFrom.Format(time.RFC3339Nano), timeTo.Format(time.RFC3339Nano))
-		entriesArray := database.GetEntriesFromDb(timeFrom, timeTo, &protocolFilter)
+		timestampTo := time.Now().UnixNano() / int64(time.Millisecond)
+		logger.Log.Infof("Getting entries from %v, to %v\n", timestampFrom, timestampTo)
+		protocolFilter := "http"
+		entriesArray := database.GetEntriesFromDb(timestampFrom, timestampTo, &protocolFilter)
 
 		if len(entriesArray) > 0 {
 			result := make([]har.Entry, 0)
@@ -276,14 +276,13 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 			analyzeInformation.SentCount += len(entriesArray)
 			logger.Log.Infof("Finish uploading %v entries to %s\n", len(entriesArray), GetTrafficDumpUrl(envPrefix, model))
 
-			logger.Log.Infof("Uploaded %v entries until now", analyzeInformation.SentCount)
 		} else {
 			logger.Log.Infof("Nothing to upload")
 		}
 
 		logger.Log.Infof("Sleeping for %v...\n", sleepTime)
 		time.Sleep(sleepTime)
-		timeFrom = timeTo
+		timestampFrom = timestampTo
 	}
 }
 

cli/cmd/clean.go

@@ -1,20 +0,0 @@
-package cmd
-
-import (
-	"github.com/spf13/cobra"
-	"github.com/up9inc/mizu/cli/telemetry"
-)
-
-var cleanCmd = &cobra.Command{
-	Use:   "clean",
-	Short: "Removes all mizu resources",
-	RunE: func(cmd *cobra.Command, args []string) error {
-		go telemetry.ReportRun("clean", nil)
-		performCleanCommand()
-		return nil
-	},
-}
-
-func init() {
-	rootCmd.AddCommand(cleanCmd)
-}

cli/cmd/cleanRunner.go

@@ -1,17 +0,0 @@
-package cmd
-
-import (
-	"github.com/up9inc/mizu/cli/config"
-	"github.com/up9inc/mizu/cli/kubernetes"
-	"github.com/up9inc/mizu/shared/logger"
-)
-
-func performCleanCommand() {
-	kubernetesProvider, err := kubernetes.NewProvider(config.Config.KubeConfigPath())
-	if err != nil {
-		logger.Log.Error(err)
-		return
-	}
-
-	finishMizuExecution(kubernetesProvider)
-}

cli/cmd/tapRunner.go

@@ -2,11 +2,8 @@ package cmd
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io/ioutil"
-	k8serrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"path"
 	"regexp"
 	"strings"
@@ -132,18 +129,11 @@ func RunMizuTap() {
 		return
 	}
 
+	defer finishMizuExecution(kubernetesProvider)
 	if err := createMizuResources(ctx, kubernetesProvider, mizuValidationRules, contract); err != nil {
 		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error creating resources: %v", errormessage.FormatError(err)))
-
-		var statusError *k8serrors.StatusError
-		if errors.As(err, &statusError) {
-			if statusError.ErrStatus.Reason == metav1.StatusReasonAlreadyExists {
-				logger.Log.Info("Mizu is already running in this namespace, change the `mizu-resources-namespace` configuration or run `mizu clean` to remove the currently running Mizu instance")
-			}
-		}
 		return
 	}
-	defer finishMizuExecution(kubernetesProvider)
 
 	go goUtils.HandleExcWrapper(watchApiServerPod, ctx, kubernetesProvider, cancel, mizuApiFilteringOptions)
 	go goUtils.HandleExcWrapper(watchTapperPod, ctx, kubernetesProvider, cancel)

cli/config/configStruct.go

@@ -14,7 +14,6 @@ import (
 const (
 	MizuResourcesNamespaceConfigName = "mizu-resources-namespace"
 	ConfigFilePathCommandName        = "config-path"
-	KubeConfigPathConfigName         = "kube-config-path"
 )
 
 type ConfigStruct struct {

cli/kubernetes/provider.go

@@ -7,9 +7,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/up9inc/mizu/shared/semver"
-	"k8s.io/apimachinery/pkg/version"
-	"net/url"
 	"path/filepath"
 	"regexp"
 	"strconv"
@@ -80,14 +77,6 @@ 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
-	}
-
-	if err := validateKubernetesVersion(clientSet); err != nil {
-		return nil, err
-	}
-
 	return &Provider{
 		clientSet:        clientSet,
 		kubernetesConfig: kubernetesConfig,
@@ -738,47 +727,3 @@ 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
-}
-
-func validateKubernetesVersion(clientSet *kubernetes.Clientset) error {
-	serverVersion, err := clientSet.ServerVersion()
-	if err != nil {
-		logger.Log.Debugf("error while getting kubernetes server version, err: %v", err)
-		return nil
-	}
-
-	serverVersionSemVer := semver.SemVersion(serverVersion.GitVersion)
-	minKubernetesServerVersionSemVer := semver.SemVersion(mizu.MinKubernetesServerVersion)
-	if minKubernetesServerVersionSemVer.GreaterThan(serverVersionSemVer) {
-		return fmt.Errorf("kubernetes server version %v is not supported, supporting only kubernetes server version of %v or higher", serverVersion.GitVersion, mizu.MinKubernetesServerVersion)
-	}
-
-	return nil
-}

cli/mizu/consts.go

@@ -14,18 +14,17 @@ var (
 )
 
 const (
-	MizuResourcesPrefix        = "mizu-"
-	ApiServerPodName           = MizuResourcesPrefix + "api-server"
-	ClusterRoleBindingName     = MizuResourcesPrefix + "cluster-role-binding"
-	ClusterRoleName            = MizuResourcesPrefix + "cluster-role"
-	K8sAllNamespaces           = ""
-	RoleBindingName            = MizuResourcesPrefix + "role-binding"
-	RoleName                   = MizuResourcesPrefix + "role"
-	ServiceAccountName         = MizuResourcesPrefix + "service-account"
-	TapperDaemonSetName        = MizuResourcesPrefix + "tapper-daemon-set"
-	TapperPodName              = MizuResourcesPrefix + "tapper"
-	ConfigMapName              = MizuResourcesPrefix + "policy"
-	MinKubernetesServerVersion = "1.16.0"
+	MizuResourcesPrefix    = "mizu-"
+	ApiServerPodName       = MizuResourcesPrefix + "api-server"
+	ClusterRoleBindingName = MizuResourcesPrefix + "cluster-role-binding"
+	ClusterRoleName        = MizuResourcesPrefix + "cluster-role"
+	K8sAllNamespaces       = ""
+	RoleBindingName        = MizuResourcesPrefix + "role-binding"
+	RoleName               = MizuResourcesPrefix + "role"
+	ServiceAccountName     = MizuResourcesPrefix + "service-account"
+	TapperDaemonSetName    = MizuResourcesPrefix + "tapper-daemon-set"
+	TapperPodName          = MizuResourcesPrefix + "tapper"
+	ConfigMapName          = MizuResourcesPrefix + "policy"
 )
 
 func GetMizuFolderPath() string {

debug.Dockerfile

@@ -12,7 +12,7 @@ FROM golang:1.16-alpine AS builder
 # Set necessary environment variables needed for our image.
 ENV CGO_ENABLED=1 GOOS=linux GOARCH=amd64
 
-RUN apk add libpcap-dev gcc g++ make bash
+RUN apk add libpcap-dev gcc g++ make
 
 # Move to agent working directory (/agent-build).
 WORKDIR /app/agent-build
@@ -20,25 +20,17 @@ WORKDIR /app/agent-build
 COPY agent/go.mod agent/go.sum ./
 COPY shared/go.mod shared/go.mod ../shared/
 COPY tap/go.mod tap/go.mod ../tap/
-COPY tap/api/go.* ../tap/api/
+
 RUN go mod download
 # cheap trick to make the build faster (As long as go.mod wasn't changes)
 RUN go list -f '{{.Path}}@{{.Version}}' -m all | sed 1d | grep -e 'go-cache' -e 'sqlite' | xargs go get
 
-ARG COMMIT_HASH
-ARG GIT_BRANCH
-ARG BUILD_TIMESTAMP
-ARG SEM_VER
-
 # Copy and build agent code
 COPY shared ../shared
 COPY tap ../tap
 COPY agent .
 RUN go build -gcflags="all=-N -l" -o mizuagent .
 
-COPY devops/build_extensions_debug.sh ..
-RUN cd .. && /bin/bash build_extensions_debug.sh
-
 
 FROM golang:1.16-alpine
 
@@ -47,7 +39,6 @@ WORKDIR /app
 
 # Copy binary and config files from /build to root folder of scratch container.
 COPY --from=builder ["/app/agent-build/mizuagent", "."]
-COPY --from=builder ["/app/agent/build/extensions", "extensions"]
 COPY --from=site-build ["/app/ui-build/build", "site"]
 
 # install remote debugging tool

devops/build-push-featurebranch-debug.sh

@@ -1,28 +0,0 @@
-#!/bin/bash
-set -e
-
-GCP_PROJECT=up9-docker-hub
-REPOSITORY=gcr.io/$GCP_PROJECT
-SERVER_NAME=mizu
-GIT_BRANCH=$(git branch | grep \* | cut -d ' ' -f2 | tr '[:upper:]' '[:lower:]')
-
-DOCKER_REPO=$REPOSITORY/$SERVER_NAME/$GIT_BRANCH
-SEM_VER=${SEM_VER=0.0.0}
-
-DOCKER_TAGGED_BUILDS=("$DOCKER_REPO:latest" "$DOCKER_REPO:$SEM_VER")
-
-if [ "$GIT_BRANCH" = 'develop' -o "$GIT_BRANCH" = 'master' -o "$GIT_BRANCH" = 'main' ]
-then
-  echo "Pushing to $GIT_BRANCH is allowed only via CI"
-  exit 1
-fi
-
-echo "building ${DOCKER_TAGGED_BUILDS[@]}"
-DOCKER_TAGS_ARGS=$(echo ${DOCKER_TAGGED_BUILDS[@]/#/-t }) # "-t FIRST_TAG -t SECOND_TAG ..."
-docker build -f debug.Dockerfile $DOCKER_TAGS_ARGS --build-arg SEM_VER=${SEM_VER} --build-arg BUILD_TIMESTAMP=${BUILD_TIMESTAMP} --build-arg GIT_BRANCH=${GIT_BRANCH} --build-arg COMMIT_HASH=${COMMIT_HASH} .
-
-for DOCKER_TAG in "${DOCKER_TAGGED_BUILDS[@]}"
-do
-  echo pushing "$DOCKER_TAG"
-  docker push "$DOCKER_TAG"
-done

devops/build_extensions_debug.sh

@@ -1,12 +0,0 @@
-#!/bin/bash
-
-for f in tap/extensions/*; do
-    if [ -d "$f" ]; then
-        extension=$(basename $f) && \
-        cd tap/extensions/${extension} && \
-        go build -gcflags="all=-N -l" -buildmode=plugin -o ../${extension}.so .  && \
-        cd ../../..  && \
-        mkdir -p agent/build/extensions  && \
-        cp tap/extensions/${extension}.so agent/build/extensions
-    fi
-done

tap/diagnose/diagnose.go

@@ -1,76 +0,0 @@
-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
-}

tap/diagnose/internal_stats.go

@@ -1,46 +0,0 @@
-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)
-}

tap/errors_map.go

@@ -1,41 +1,23 @@
-package diagnose
+package tap
 
 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
+	outputLevel    int
+	nErrors        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 {
+func NewErrorsMap(outputLevel int) *errorsMap {
 	return &errorsMap{
 		errorsMap:   make(map[string]uint),
-		OutputLevel: outputLevel,
+		outputLevel: outputLevel,
 	}
 }
 
@@ -46,12 +28,12 @@ func newErrorsMap(outputLevel int) *errorsMap {
  */
 func (e *errorsMap) logError(minOutputLevel int, t string, s string, a ...interface{}) {
 	e.errorsMapMutex.Lock()
-	e.ErrorsCount++
+	e.nErrors++
 	nb := e.errorsMap[t]
 	e.errorsMap[t] = nb + 1
 	e.errorsMapMutex.Unlock()
 
-	if e.OutputLevel >= minOutputLevel {
+	if e.outputLevel >= minOutputLevel {
 		formatStr := fmt.Sprintf("%s: %s", t, s)
 		logger.Log.Errorf(formatStr, a...)
 	}
@@ -69,17 +51,10 @@ func (e *errorsMap) Debug(s string, a ...interface{}) {
 	logger.Log.Debugf(s, a...)
 }
 
-func (e *errorsMap) GetErrorsSummary() (int, string) {
+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])
-	}
-}

tap/net_utils.go

@@ -3,8 +3,6 @@ package tap
 import (
 	"net"
 	"strings"
-
-	"github.com/up9inc/mizu/tap/diagnose"
 )
 
 var privateIPBlocks []*net.IPNet
@@ -57,7 +55,7 @@ func initPrivateIPBlocks() {
 	} {
 		_, block, err := net.ParseCIDR(cidr)
 		if err != nil {
-			diagnose.TapErrors.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/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,
 	}
 }
 

tap/passive_tapper.go

@@ -9,17 +9,28 @@
 package tap
 
 import (
+	"encoding/hex"
 	"encoding/json"
 	"flag"
+	"fmt"
+	"io"
 	"os"
+	"os/signal"
 	"runtime"
+	"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
@@ -51,6 +62,28 @@ 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 {
+	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
 }
@@ -77,121 +110,353 @@ 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() {
-		diagnose.StartMemoryProfiler(os.Getenv(MemoryProfilingDumpPath), os.Getenv(MemoryProfilingTimeIntervalSeconds))
+		startMemoryProfiler()
 	}
 
 	go startPassiveTapper(outputItems)
 }
 
-func printPeriodicStats(cleaner *Cleaner) {
-	statsPeriod := time.Second * time.Duration(*statsevery)
-	ticker := time.NewTicker(statsPeriod)
-
-	for {
-		<-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 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
+		}
+	}
+
+	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 startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 	streamsMap := NewTcpStreamMap()
 	go streamsMap.closeTimedoutTcpStreamChannels()
 
-	diagnose.InitializeErrorsMap(*debug, *verbose, *quiet)
-	diagnose.InitializeTapperInternalStats()
+	var outputLevel int
 
-	var bpffilter string
+	defer util.Run()()
+	if *debug {
+		outputLevel = 2
+	} else if *verbose {
+		outputLevel = 1
+	} else if *quiet {
+		outputLevel = -1
+	}
+
+	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)
+		}
+	} 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()
+	}
 	if len(flag.Args()) > 0 {
-		bpffilter = strings.Join(flag.Args(), " ")
+		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)
+		}
 	}
 
-	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)
+	var dec gopacket.Decoder
+	var ok bool
+	decoderName := *decoder
+	if decoderName == "" {
+		decoderName = handle.LinkType().String()
 	}
-
-	defer packetSource.Close()
-
-	if err != nil {
-		logger.Log.Fatal(err)
+	if dec, ok = gopacket.DecodersByLayerName[decoderName]; !ok {
+		logger.Log.Fatal("No decoder named", decoderName)
 	}
-
-	packets := make(chan source.TcpPacketInfo)
-	assembler := NewTcpAssembler(outputItems, streamsMap)
-
+	source := gopacket.NewPacketSource(handle, dec)
+	source.Lazy = *lazy
+	source.NoCopy = true
 	logger.Log.Info("Starting to read packets")
-	diagnose.AppStats.SetStartTime(time.Now())
+	appStats.SetStartTime(time.Now())
+	defragger := ip4defrag.NewIPv4Defragmenter()
 
-	go packetSource.ReadPackets(!*nodefrag, packets)
+	var emitter api.Emitter = &api.Emitting{
+		AppStats:      &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
+
+	var assemblerMutex sync.Mutex
+
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, os.Interrupt)
 
 	staleConnectionTimeout := time.Second * time.Duration(*staleTimeoutSeconds)
 	cleaner := Cleaner{
-		assembler:         assembler.Assembler,
-		assemblerMutex:    &assembler.assemblerMutex,
+		assembler:         assembler,
+		assemblerMutex:    &assemblerMutex,
 		cleanPeriod:       cleanPeriod,
 		connectionTimeout: staleConnectionTimeout,
 	}
 	cleaner.start()
 
-	go printPeriodicStats(&cleaner)
+	go func() {
+		statsPeriod := time.Second * time.Duration(*statsevery)
+		ticker := time.NewTicker(statsPeriod)
 
-	assembler.processPackets(*hexdumppkt, packets)
+		for {
+			<-ticker.C
 
-	if diagnose.TapErrors.OutputLevel >= 2 {
-		assembler.dumpStreamPool()
+			// Since the start
+			errorMapLen, errorsSummery := tapErrors.getErrorsSummary()
+
+			logger.Log.Infof("%v (errors: %v, errTypes:%v) - Errors Summary: %s",
+				time.Since(appStats.StartTime),
+				tapErrors.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 err := diagnose.DumpMemoryProfile(*memprofile); err != nil {
-		logger.Log.Errorf("Error dumping memory profile %v\n", err)
+	for {
+		packet, err := source.NextPacket()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			if err.Error() != "Timeout Expired" {
+				logger.Log.Debugf("Error: %T", 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 {
+			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),
+				tapErrors.nErrors,
+				errorMapLen)
+		}
+		select {
+		case <-signalChan:
+			logger.Log.Infof("Caught SIGINT: aborting")
+			done = true
+		default:
+			// NOP: continue
+		}
+		if done {
+			break
+		}
 	}
 
-	assembler.waitAndDump()
+	assemblerMutex.Lock()
+	closed := assembler.FlushAll()
+	assemblerMutex.Unlock()
+	logger.Log.Debugf("Final flush: %d closed", closed)
+	if outputLevel >= 2 {
+		streamPool.Dump()
+	}
 
-	diagnose.InternalStats.PrintStatsSummary()
-	diagnose.TapErrors.PrintSummary()
-	logger.Log.Infof("AppStats: %v", diagnose.AppStats)
+	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", 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/source/tcp_packet_source.go

@@ -1,150 +0,0 @@
-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,
-		}
-	}
-}

tap/tcp_assembler.go

@@ -1,132 +0,0 @@
-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()
-}

tap/tcp_stream.go

@@ -9,7 +9,6 @@ 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)
@@ -37,11 +36,11 @@ type tcpStream struct {
 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) {
-		diagnose.TapErrors.SilentError("FSM-rejection", "%s: Packet rejected by FSM (state:%s)", t.ident, t.tcpstate.String())
-		diagnose.InternalStats.RejectFsm++
+		tapErrors.SilentError("FSM-rejection", "%s: Packet rejected by FSM (state:%s)", t.ident, t.tcpstate.String())
+		stats.rejectFsm++
 		if !t.fsmerr {
 			t.fsmerr = true
-			diagnose.InternalStats.RejectConnFsm++
+			stats.rejectConnFsm++
 		}
 		if !*ignorefsmerr {
 			return false
@@ -50,8 +49,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 {
-		diagnose.TapErrors.SilentError("OptionChecker-rejection", "%s: Packet rejected by OptionChecker: %s", t.ident, err)
-		diagnose.InternalStats.RejectOpt++
+		tapErrors.SilentError("OptionChecker-rejection", "%s: Packet rejected by OptionChecker: %s", t.ident, err)
+		stats.rejectOpt++
 		if !*nooptcheck {
 			return false
 		}
@@ -61,15 +60,15 @@ func (t *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 	if *checksum {
 		c, err := tcp.ComputeChecksum()
 		if err != nil {
-			diagnose.TapErrors.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 {
-			diagnose.TapErrors.SilentError("Checksum", "%s: Invalid checksum: 0x%x", t.ident, c)
+			tapErrors.SilentError("Checksum", "%s: Invalid checksum: 0x%x", t.ident, c)
 			accept = false
 		}
 	}
 	if !accept {
-		diagnose.InternalStats.RejectOpt++
+		stats.rejectOpt++
 	}
 	return accept
 }
@@ -80,28 +79,28 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 	// update stats
 	sgStats := sg.Stats()
 	if skip > 0 {
-		diagnose.InternalStats.MissedBytes += skip
+		stats.missedBytes += skip
 	}
-	diagnose.InternalStats.Sz += length - saved
-	diagnose.InternalStats.Pkt += sgStats.Packets
+	stats.sz += length - saved
+	stats.pkt += sgStats.Packets
 	if sgStats.Chunks > 1 {
-		diagnose.InternalStats.Reassembled++
+		stats.reassembled++
 	}
-	diagnose.InternalStats.OutOfOrderPackets += sgStats.QueuedPackets
-	diagnose.InternalStats.OutOfOrderBytes += sgStats.QueuedBytes
-	if length > diagnose.InternalStats.BiggestChunkBytes {
-		diagnose.InternalStats.BiggestChunkBytes = length
+	stats.outOfOrderPackets += sgStats.QueuedPackets
+	stats.outOfOrderBytes += sgStats.QueuedBytes
+	if length > stats.biggestChunkBytes {
+		stats.biggestChunkBytes = length
 	}
-	if sgStats.Packets > diagnose.InternalStats.BiggestChunkPackets {
-		diagnose.InternalStats.BiggestChunkPackets = sgStats.Packets
+	if sgStats.Packets > stats.biggestChunkPackets {
+		stats.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.
-		diagnose.TapErrors.SilentError("Invalid-Overlap", "bytes:%d, pkts:%d", sgStats.OverlapBytes, sgStats.OverlapPackets)
+		tapErrors.SilentError("Invalid-Overlap", "bytes:%d, pkts:%d", sgStats.OverlapBytes, sgStats.OverlapPackets)
 	}
-	diagnose.InternalStats.OverlapBytes += sgStats.OverlapBytes
-	diagnose.InternalStats.OverlapPackets += sgStats.OverlapPackets
+	stats.overlapBytes += sgStats.OverlapBytes
+	stats.overlapPackets += sgStats.OverlapPackets
 
 	var ident string
 	if dir == reassembly.TCPDirClientToServer {
@@ -109,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)
 	}
-	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)
+	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 {
@@ -128,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:])
-		diagnose.TapErrors.Debug("dnsSize: %d, missing: %d", dnsSize, missing)
+		tapErrors.Debug("dnsSize: %d, missing: %d", dnsSize, missing)
 		if missing > 0 {
-			diagnose.TapErrors.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 {
-			diagnose.TapErrors.SilentError("DNS-parser", "Failed to decode DNS: %v", err)
+			tapErrors.SilentError("DNS-parser", "Failed to decode DNS: %v", err)
 		} else {
-			diagnose.TapErrors.Debug("DNS: %s", gopacket.LayerDump(dns))
+			tapErrors.Debug("DNS: %s", gopacket.LayerDump(dns))
 		}
 		if len(data) > 2+int(dnsSize) {
 			sg.KeepFrom(2 + int(dnsSize))
@@ -148,7 +147,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
-			diagnose.AppStats.IncReassembledTcpPayloadsCount()
+			appStats.IncReassembledTcpPayloadsCount()
 			timestamp := ac.GetCaptureInfo().Timestamp
 			if dir == reassembly.TCPDirClientToServer {
 				for i := range t.clients {
@@ -174,7 +173,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 }
 
 func (t *tcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
-	diagnose.TapErrors.Debug("%s: Connection closed", t.ident)
+	tapErrors.Debug("%s: Connection closed", t.ident)
 	if t.isTapTarget && !t.isClosed {
 		t.Close()
 	}

tap/tcp_streams_map.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/up9inc/mizu/shared/logger"
-	"github.com/up9inc/mizu/tap/diagnose"
 )
 
 type tcpStreamMap struct {
@@ -45,9 +44,9 @@ func (streamMap *tcpStreamMap) closeTimedoutTcpStreamChannels() {
 			if stream.superIdentifier.Protocol == nil {
 				if !stream.isClosed && time.Now().After(streamWrapper.createdAt.Add(tcpStreamChannelTimeout)) {
 					stream.Close()
-					diagnose.AppStats.IncDroppedTcpStreams()
+					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)
+						appStats.DroppedTcpStreams, runtime.NumGoroutine(), tcpStreamChannelTimeout/1000000)
 				}
 			} else {
 				if !stream.superIdentifier.IsClosedOthers {

tap/tester/tester.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path"
@@ -33,7 +34,7 @@ func loadExtensions() ([]*tapApi.Extension, error) {
 			continue
 		}
 
-		logger.Log.Infof("Loading extension: %s\n", filename)
+		fmt.Printf("Loading extension: %s\n", filename)
 
 		extension := &tapApi.Extension{
 			Path: path.Join(extensionsDir, filename),
@@ -68,7 +69,7 @@ func loadExtensions() ([]*tapApi.Extension, error) {
 	})
 
 	for _, extension := range extensions {
-		logger.Log.Infof("Extension Properties: %+v\n", extension)
+		fmt.Printf("Extension Properties: %+v\n", extension)
 	}
 
 	return extensions, nil
@@ -92,7 +93,7 @@ func internalRun() error {
 
 	tap.StartPassiveTapper(&opts, outputItems, extenssions, &tapOpts)
 
-	logger.Log.Infof("Tapping, press enter to exit...\n")
+	fmt.Printf("Tapping, press enter to exit...\n")
 	reader := bufio.NewReader(os.Stdin)
 	reader.ReadLine()
 	return nil
@@ -104,9 +105,9 @@ func main() {
 	if err != nil {
 		switch err := err.(type) {
 		case *errors.Error:
-			logger.Log.Errorf("Error: %v\n", err.ErrorStack())
+			fmt.Printf("Error: %v\n", err.ErrorStack())
 		default:
-			logger.Log.Errorf("Error: %v\n", err)
+			fmt.Printf("Error: %v\n", err)
 		}
 
 		os.Exit(1)