fixed upload entries (#407)

* debug mizu

* Update clean.go, cleanRunner.go, and tapRunner.go

* Update tapRunner.go

* Update cleanRunner.go

* Update tapRunner.go

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

Makefile

@@ -46,6 +46,10 @@ 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,11 +13,12 @@ import (
 )
 
 const (
-	DBPath    = "./entries.db"
-	OrderDesc = "desc"
-	OrderAsc  = "asc"
-	LT        = "lt"
-	GT        = "gt"
+	DBPath     = "./entries.db"
+	OrderDesc  = "desc"
+	OrderAsc   = "asc"
+	LT         = "lt"
+	GT         = "gt"
+	TimeFormat = "2006-01-02 15:04:05.000000000"
 )
 
 var (
@@ -57,7 +58,7 @@ func initDataBase(databasePath string) *gorm.DB {
 	return temp
 }
 
-func GetEntriesFromDb(timestampFrom int64, timestampTo int64, protocolName *string) []tapApi.MizuEntry {
+func GetEntriesFromDb(timeFrom time.Time, timeTo time.Time, protocolName *string) []tapApi.MizuEntry {
 	order := OrderDesc
 	protocolNameCondition := "1 = 1"
 	if protocolName != nil {
@@ -67,7 +68,7 @@ func GetEntriesFromDb(timestampFrom int64, timestampTo int64, protocolName *stri
 	var entries []tapApi.MizuEntry
 	GetEntriesTable().
 		Where(protocolNameCondition).
-		Where(fmt.Sprintf("timestamp BETWEEN %v AND %v", timestampFrom, timestampTo)).
+		Where(fmt.Sprintf("created_at BETWEEN '%s' AND '%s'", timeFrom.Format(TimeFormat), timeTo.Format(TimeFormat))).
 		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 timestampFrom int64 = 0
+	var timeFrom time.Time
+	protocolFilter := "http"
 
 	for {
-		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)
+		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)
 
 		if len(entriesArray) > 0 {
 			result := make([]har.Entry, 0)
@@ -276,13 +276,14 @@ 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)
-		timestampFrom = timestampTo
+		timeFrom = timeTo
 	}
 }
 

cli/cmd/clean.go

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

@@ -0,0 +1,17 @@
+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,8 +2,11 @@ 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"
@@ -129,11 +132,18 @@ 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/kubernetes/provider.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/up9inc/mizu/shared/semver"
 	"k8s.io/apimachinery/pkg/version"
 	"net/url"
 	"path/filepath"
@@ -83,6 +84,10 @@ func NewProvider(kubeConfigPath string) (*Provider, error) {
 		return nil, err
 	}
 
+	if err := validateKubernetesVersion(clientSet); err != nil {
+		return nil, err
+	}
+
 	return &Provider{
 		clientSet:        clientSet,
 		kubernetesConfig: kubernetesConfig,
@@ -761,3 +766,19 @@ func validateNotProxy(kubernetesConfig clientcmd.ClientConfig, restClientConfig
 
 	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,17 +14,18 @@ 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"
+	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"
 )
 
 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
+RUN apk add libpcap-dev gcc g++ make bash
 
 # Move to agent working directory (/agent-build).
 WORKDIR /app/agent-build
@@ -20,17 +20,25 @@ 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
 
@@ -39,6 +47,7 @@ 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

@@ -0,0 +1,28 @@
+#!/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

@@ -0,0 +1,12 @@
+#!/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/tcp_packet_source.go

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