From da24608bec189a0af1c43c5a3bfa97d344075054 Mon Sep 17 00:00:00 2001 From: nimrod-up9 <59927337+nimrod-up9@users.noreply.github.com> Date: Thu, 20 May 2021 12:22:23 +0300 Subject: [PATCH] Tap multiple pods statically (#51) * WIP * Update tap.go, provider.go, and 2 more files... * WIP * WIP * Solved routine hanging forever: Added missing flag when calling mizuagent. * Iterate channel with range. * Panic if har channel is nil or if websocket connection is nil. * StartPassiveTapper returns read only channel. * Solved program exiting immediately: Wait for interrupt signal instead of exiting. * Solve connecting issue - Retry a few times. * Use lib const instead of magic. * Nicer error prints. * Don't coninue piping message if there is an error. * Comment. * Dependency injection. * no message * Fixed comment. * Print tapped addresses when they are updated. * Print errors in cleanup if there are any. Co-authored-by: RamiBerm Co-authored-by: Roee Gadot --- Dockerfile | 2 +- api/go.mod | 32 +----- api/go.sum | 11 -- api/main.go | 72 +++++++++++-- api/pkg/{inserter => api}/main.go | 14 +-- api/pkg/api/socket_client.go | 60 +++++++++++ api/pkg/api/socket_server_handlers.go | 76 ++++++++++++++ api/pkg/routes/socket_routes.go | 48 ++++----- api/pkg/tap/passive_tapper.go | 9 +- api/pkg/tap/tcp_stream_factory.go | 4 +- cli/cmd/tap.go | 15 ++- cli/kubernetes/provider.go | 144 +++++++++++++++++++++----- cli/mizu/consts.go | 3 + cli/mizu/mizuRunner.go | 133 ++++++++++++++++-------- 14 files changed, 461 insertions(+), 162 deletions(-) rename api/pkg/{inserter => api}/main.go (91%) create mode 100644 api/pkg/api/socket_client.go create mode 100644 api/pkg/api/socket_server_handlers.go diff --git a/Dockerfile b/Dockerfile index e442cd60f..bc276a398 100644 --- a/Dockerfile +++ b/Dockerfile @@ -38,4 +38,4 @@ COPY --from=site-build ["/ui-build/build", "site"] COPY api/start.sh . # this script runs both apiserver and passivetapper and exits either if one of them exits, preventing a scenario where the container runs without one process -CMD "./start.sh" +ENTRYPOINT "/app/mizuagent" diff --git a/api/go.mod b/api/go.mod index 3069f5302..d0d831ebc 100644 --- a/api/go.mod +++ b/api/go.mod @@ -3,37 +3,8 @@ module mizuserver go 1.16 require ( - cloud.google.com/go v0.54.0 // indirect - cloud.google.com/go/bigquery v1.4.0 // indirect - cloud.google.com/go/datastore v1.1.0 // indirect - cloud.google.com/go/pubsub v1.2.0 // indirect - cloud.google.com/go/storage v1.6.0 // indirect - github.com/Azure/go-autorest v14.2.0+incompatible // indirect - github.com/Azure/go-autorest/autorest v0.11.12 // indirect - github.com/Azure/go-autorest/autorest/adal v0.9.5 // indirect - github.com/Azure/go-autorest/autorest/date v0.3.0 // indirect - github.com/Azure/go-autorest/autorest/mocks v0.4.1 // indirect - github.com/Azure/go-autorest/logger v0.2.0 // indirect - github.com/Azure/go-autorest/tracing v0.6.0 // indirect - github.com/BurntSushi/toml v0.3.1 // indirect - github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802 // indirect - github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46 // indirect - github.com/PuerkitoBio/purell v1.1.1 // indirect - github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect - github.com/andybalholm/brotli v1.0.1 // indirect github.com/antoniodipinto/ikisocket v0.0.0-20210417133349-f1502512d69a - github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a // indirect - github.com/aws/aws-sdk-go v1.34.28 // indirect - github.com/census-instrumentation/opencensus-proto v0.2.1 // indirect - github.com/chzyer/logex v1.1.10 // indirect - github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e // indirect - github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1 // indirect - github.com/client9/misspell v0.3.4 // indirect - github.com/creack/pty v1.1.9 // indirect - github.com/davecgh/go-spew v1.1.1 // indirect github.com/djherbis/atime v1.0.0 - github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815 // indirect - github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153 // indirect github.com/fasthttp/websocket v1.4.3-beta.1 // indirect github.com/go-playground/locales v0.13.0 github.com/go-playground/universal-translator v0.17.0 @@ -45,12 +16,11 @@ require ( github.com/leodido/go-urn v1.2.1 // indirect github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 github.com/patrickmn/go-cache v2.1.0+incompatible - github.com/stretchr/objx v0.2.0 // indirect go.mongodb.org/mongo-driver v1.5.1 golang.org/x/net v0.0.0-20210421230115-4e50805a0758 gorm.io/driver/sqlite v1.1.4 gorm.io/gorm v1.21.8 - k8s.io/api v0.21.0 // indirect + k8s.io/api v0.21.0 k8s.io/apimachinery v0.21.0 k8s.io/client-go v0.21.0 ) diff --git a/api/go.sum b/api/go.sum index 2a30e989f..00d32ca00 100644 --- a/api/go.sum +++ b/api/go.sum @@ -281,15 +281,12 @@ github.com/klauspost/cpuid v1.2.1/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgo github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s= github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= -github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs= github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= -github.com/kr/pty v1.1.1 h1:VkoXIwSboBpnk99O/KFauAEILuNHv5DVFKZMBN/gUgw= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/pty v1.1.5 h1:hyz3dwM5QLc1Rfoz4FuWJQG5BN7tc6K1MndAUnGpQr4= github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA= -github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= @@ -362,7 +359,6 @@ github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTd github.com/spf13/cobra v0.0.3 h1:ZlrZ4XsMRm04Fr5pSFxBgfND2EBVa1nLpiy1stUsX/8= github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ= github.com/spf13/pflag v0.0.0-20170130214245-9ff6c6923cff/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= -github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg= github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= @@ -448,7 +444,6 @@ golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028 h1:4+4C/Iv2U4fMZBiMCc98MG golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028/go.mod h1:E/iHnbuqvinMTCcRqshq8CkpyQDoeVncDDYHnLhea+o= golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc= golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY= -golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee h1:WG0RUwxtNT4qqaXX3DPA8zHFNm/D9xaBpxzHt1WcA/E= golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= @@ -494,7 +489,6 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190412183630-56d357773e84/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -534,7 +528,6 @@ golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210420072515-93ed5bcd2bfe h1:WdX7u8s3yOigWAhHEaDl8r9G+4XwFQEQFtBMYyN+kXQ= golang.org/x/sys v0.0.0-20210420072515-93ed5bcd2bfe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= -golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d h1:SZxvLBoTP5yHO3Frd4z4vrF+DBX9vMVanchswa69toE= golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= @@ -581,7 +574,6 @@ golang.org/x/tools v0.0.0-20191216173652-a0e659d51361/go.mod h1:TB2adYChydJhpapK golang.org/x/tools v0.0.0-20191227053925-7b8e75db28f4/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.0.0-20200117161641-43d50277825c/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.0.0-20200122220014-bf1340f18c4a/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= -golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7 h1:EBZoQjiKKPaLbPrbpssUfuHtwM6KV/vb4U85g/cigFY= golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.0.0-20200204074204-1cc6d1ef6c74/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.0.0-20200207183749-b753a1ba74fa/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= @@ -593,7 +585,6 @@ golang.org/x/tools v0.0.0-20210106214847-113979e3529a h1:CB3a9Nez8M13wwlr/E2Ytwo golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -653,7 +644,6 @@ google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpAD google.golang.org/protobuf v1.25.0 h1:Ejskq+SyPohKW+1uil0JJMtmHCgJPJ/qWTxr8qp+R4c= google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU= @@ -668,7 +658,6 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkep gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10= gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= diff --git a/api/main.go b/api/main.go index e8d4242f1..2b2ce8aee 100644 --- a/api/main.go +++ b/api/main.go @@ -1,24 +1,71 @@ package main import ( + "encoding/json" "flag" + "fmt" "github.com/gofiber/fiber/v2" - "mizuserver/pkg/inserter" + "mizuserver/pkg/api" "mizuserver/pkg/middleware" "mizuserver/pkg/routes" "mizuserver/pkg/tap" "mizuserver/pkg/utils" + "os" + "os/signal" ) +var shouldTap = flag.Bool("tap", false, "Run in tapper mode without API") +var aggregator = flag.Bool("aggregator", false, "Run in aggregator mode with API") +var standalone = flag.Bool("standalone", false, "Run in standalone tapper and API mode") +var aggregatorAddress = flag.String("aggregator-address", "", "Address of mizu collector for tapping") + +const nodeNameEnvVar = "NODE_NAME" +const tappedAddressesPerNodeDictEnvVar = "TAPPED_ADDRESSES_PER_HOST" + func main() { flag.Parse() - harOutputChannel := tap.StartPassiveTapper() + if !*shouldTap && !*aggregator && !*standalone{ + panic("One of the flags --tap, --api or --standalone must be provided") + } + if *standalone { + harOutputChannel := tap.StartPassiveTapper() + go api.StartReadingEntries(harOutputChannel, tap.HarOutputDir) + hostApi(nil) + } else if *shouldTap { + if *aggregatorAddress == "" { + panic("Aggregator address must be provided with --aggregator-address when using --tap") + } + + tapTargets := getTapTargets() + if tapTargets != nil { + tap.HostAppAddresses = tapTargets + fmt.Println("Filtering for the following addresses:", tap.HostAppAddresses) + } + + harOutputChannel := tap.StartPassiveTapper() + socketConnection, err := api.ConnectToSocketServer(*aggregatorAddress) + if err != nil { + panic(fmt.Sprintf("Error connecting to socket server at %s %v", *aggregatorAddress, err)) + } + go api.PipeChannelToSocket(socketConnection, harOutputChannel) + } else if *aggregator { + socketHarOutChannel := make(chan *tap.OutputChannelItem, 1000) + go api.StartReadingEntries(socketHarOutChannel, nil) + hostApi(socketHarOutChannel) + } + + signalChan := make(chan os.Signal, 1) + signal.Notify(signalChan, os.Interrupt) + <-signalChan + + fmt.Println("Exiting") +} + +func hostApi(socketHarOutputChannel chan<- *tap.OutputChannelItem) { app := fiber.New() - // process to read files / channel and insert to DB - go inserter.StartReadingEntries(harOutputChannel, tap.HarOutputDir) middleware.FiberMiddleware(app) // Register Fiber's middleware for app. app.Static("/", "./site") @@ -27,10 +74,23 @@ func main() { app.Get("/echo", func(c *fiber.Ctx) error { return c.SendString("Hello, World 👋!") }) - - routes.WebSocketRoutes(app) + eventHandlers := api.RoutesEventHandlers{ + SocketHarOutChannel: socketHarOutputChannel, + } + routes.WebSocketRoutes(app, &eventHandlers) routes.EntriesRoutes(app) routes.NotFoundRoute(app) utils.StartServer(app) } + + +func getTapTargets() []string { + nodeName := os.Getenv(nodeNameEnvVar) + var tappedAddressesPerNodeDict map[string][]string + err := json.Unmarshal([]byte(os.Getenv(tappedAddressesPerNodeDictEnvVar)), &tappedAddressesPerNodeDict) + if err != nil { + panic(fmt.Sprintf("env var value of %s is invalid! must be map[string][]string %v", tappedAddressesPerNodeDict, err)) + } + return tappedAddressesPerNodeDict[nodeName] +} diff --git a/api/pkg/inserter/main.go b/api/pkg/api/main.go similarity index 91% rename from api/pkg/inserter/main.go rename to api/pkg/api/main.go index 4089fb01d..1d2882bda 100644 --- a/api/pkg/inserter/main.go +++ b/api/pkg/api/main.go @@ -1,11 +1,10 @@ -package inserter +package api import ( "bufio" "context" "encoding/json" "fmt" - "github.com/antoniodipinto/ikisocket" "github.com/google/martian/har" "go.mongodb.org/mongo-driver/bson/primitive" "mizuserver/pkg/database" @@ -43,7 +42,7 @@ func init() { k8sResolver = res } -func StartReadingEntries(harChannel chan *tap.OutputChannelItem, workingDir *string) { +func StartReadingEntries(harChannel <-chan *tap.OutputChannelItem, workingDir *string) { if workingDir != nil && *workingDir != "" { startReadingFiles(*workingDir) } else { @@ -83,7 +82,11 @@ func startReadingFiles(workingDir string) { } } -func startReadingChannel(outputItems chan *tap.OutputChannelItem) { +func startReadingChannel(outputItems <-chan *tap.OutputChannelItem) { + if outputItems == nil { + panic("Channel of captured messages is nil") + } + for item := range outputItems { saveHarToDb(item.HarEntry, item.RequestSenderIp) } @@ -118,7 +121,7 @@ func saveHarToDb(entry *har.Entry, sender string) { baseEntry := utils.GetResolvedBaseEntry(mizuEntry) baseEntryBytes, _ := json.Marshal(&baseEntry) - ikisocket.Broadcast(baseEntryBytes) + broadcastToBrowserClients(baseEntryBytes) } func getServiceNameFromUrl(inputUrl string) (string, string, string) { @@ -126,4 +129,3 @@ func getServiceNameFromUrl(inputUrl string) (string, string, string) { utils.CheckErr(err) return fmt.Sprintf("%s://%s", parsed.Scheme, parsed.Host), parsed.Path, parsed.Host } - diff --git a/api/pkg/api/socket_client.go b/api/pkg/api/socket_client.go new file mode 100644 index 000000000..db522be4c --- /dev/null +++ b/api/pkg/api/socket_client.go @@ -0,0 +1,60 @@ +package api + +import ( + "encoding/json" + "fmt" + "github.com/gorilla/websocket" + "mizuserver/pkg/tap" + "time" +) + +func ConnectToSocketServer(address string) (*websocket.Conn, error) { + const maxTry = 3 + const sleepTime = time.Second * 10 + var err error + var connection *websocket.Conn + try := 0 + + // Connection to server fails if client pod is up before server. + // Retries solve this issue. + for try < maxTry { + connection, _, err = websocket.DefaultDialer.Dial(address, nil) + if err != nil { + try++ + fmt.Printf("Failed connecting to websocket server: %s, (%v,%+v)\n", err, err, err) + } else { + break + } + time.Sleep(sleepTime) + } + + if err != nil { + return nil, err + } + + return connection, nil +} + +func PipeChannelToSocket(connection *websocket.Conn, messageDataChannel <-chan *tap.OutputChannelItem) { + if connection == nil { + panic("Websocket connection is nil") + } + + if messageDataChannel == nil { + panic("Channel of captured messages is nil") + } + + for messageData := range messageDataChannel { + marshaledData, err := json.Marshal(messageData) + if err != nil { + fmt.Printf("error converting message to json %s, (%v,%+v)\n", err, err, err) + continue + } + + err = connection.WriteMessage(websocket.TextMessage, marshaledData) + if err != nil { + fmt.Printf("error sending message through socket server %s, (%v,%+v)\n", err, err, err) + continue + } + } +} diff --git a/api/pkg/api/socket_server_handlers.go b/api/pkg/api/socket_server_handlers.go new file mode 100644 index 000000000..6855d4292 --- /dev/null +++ b/api/pkg/api/socket_server_handlers.go @@ -0,0 +1,76 @@ +package api + +import ( + "encoding/json" + "fmt" + "github.com/antoniodipinto/ikisocket" + "mizuserver/pkg/routes" + "mizuserver/pkg/tap" +) + +var browserClientSocketUUIDs = make([]string, 0) + +type RoutesEventHandlers struct { + routes.EventHandlers + SocketHarOutChannel chan<- *tap.OutputChannelItem +} + + +func (h *RoutesEventHandlers) WebSocketConnect(ep *ikisocket.EventPayload) { + if ep.Kws.GetAttribute("is_tapper") == true { + fmt.Println(fmt.Sprintf("Websocket Connection event - Tapper connected: %s", ep.SocketUUID)) + } else { + fmt.Println(fmt.Sprintf("Websocket Connection event - Browser socket connected: %s", ep.SocketUUID)) + browserClientSocketUUIDs = append(browserClientSocketUUIDs, ep.SocketUUID) + } +} + +func (h *RoutesEventHandlers) WebSocketDisconnect(ep *ikisocket.EventPayload) { + if ep.Kws.GetAttribute("is_tapper") == true { + fmt.Println(fmt.Sprintf("Disconnection event - Tapper connected: %s", ep.SocketUUID)) + } else { + fmt.Println(fmt.Sprintf("Disconnection event - Browser socket connected: %s", ep.SocketUUID)) + removeSocketUUIDFromBrowserSlice(ep.SocketUUID) + } +} + +func broadcastToBrowserClients(message []byte) { + ikisocket.EmitToList(browserClientSocketUUIDs, message) +} + +func (h *RoutesEventHandlers) WebSocketClose(ep *ikisocket.EventPayload) { + if ep.Kws.GetAttribute("is_tapper") == true { + fmt.Println(fmt.Sprintf("Websocket Close event - Tapper connected: %s", ep.SocketUUID)) + } else { + fmt.Println(fmt.Sprintf("Websocket Close event - Browser socket connected: %s", ep.SocketUUID)) + removeSocketUUIDFromBrowserSlice(ep.SocketUUID) + } +} + +func (h *RoutesEventHandlers) WebSocketError(ep *ikisocket.EventPayload) { + fmt.Println(fmt.Sprintf("Socket error - Socket uuid : %s %v", ep.SocketUUID, ep.Error)) +} + +func (h *RoutesEventHandlers) WebSocketMessage(ep *ikisocket.EventPayload) { + if ep.Kws.GetAttribute("is_tapper") == true && h.SocketHarOutChannel != nil{ + var tapOutput tap.OutputChannelItem + err := json.Unmarshal(ep.Data, &tapOutput) + if err != nil { + fmt.Printf("Could not unmarshal message received from tapper websocket %v", err) + } else { + h.SocketHarOutChannel <- &tapOutput + } + } else { + fmt.Println("Received Web socket message, unable to handle message") + } +} + +func removeSocketUUIDFromBrowserSlice(uuidToRemove string) { + newUUIDSlice := make([]string, 0, len(browserClientSocketUUIDs)) + for _, uuid := range browserClientSocketUUIDs { + if uuid != uuidToRemove { + newUUIDSlice = append(newUUIDSlice, uuid) + } + } + browserClientSocketUUIDs = newUUIDSlice +} diff --git a/api/pkg/routes/socket_routes.go b/api/pkg/routes/socket_routes.go index c912253ed..a66cc401b 100644 --- a/api/pkg/routes/socket_routes.go +++ b/api/pkg/routes/socket_routes.go @@ -1,43 +1,31 @@ package routes import ( - "fmt" "github.com/antoniodipinto/ikisocket" "github.com/gofiber/fiber/v2" ) -func webSocketConnect(ep *ikisocket.EventPayload) { - fmt.Println(fmt.Sprintf("Connection event 1 - User: %s", ep.Kws.GetStringAttribute("user_id"))) +type EventHandlers interface { + WebSocketConnect(ep *ikisocket.EventPayload) + WebSocketDisconnect(ep *ikisocket.EventPayload) + WebSocketClose(ep *ikisocket.EventPayload) + WebSocketError(ep *ikisocket.EventPayload) + WebSocketMessage(ep *ikisocket.EventPayload) } -func webSocketDisconnect(ep *ikisocket.EventPayload) { - fmt.Println(fmt.Sprintf("Disconnection event - User: %s", ep.Kws.GetStringAttribute("user_id"))) -} - -func webSocketClose(ep *ikisocket.EventPayload) { - fmt.Println(fmt.Sprintf("Close event - User: %s", ep.Kws.GetStringAttribute("user_id"))) -} - -func webSocketError(ep *ikisocket.EventPayload) { - fmt.Println(fmt.Sprintf("Error event - User: %s", ep.Kws.GetStringAttribute("user_id"))) -} - -func webSocketMessage(ep *ikisocket.EventPayload) { - fmt.Println("Web socket message") - // fmt.Println(fmt.Sprintf("Message event - User: %s - Message: %s", ep.Kws.GetStringAttribute("user_id"), string(ep.Data))) -} - -func WebSocketRoutes(app *fiber.App) { - +func WebSocketRoutes(app *fiber.App, eventHandlers EventHandlers) { app.Get("/ws", ikisocket.New(func(kws *ikisocket.Websocket) { - // kws.Broadcast([]byte(fmt.Sprintf("New user connected: %s and UUID: %s", userId, kws.UUID)), true) - // kws.Emit([]byte(fmt.Sprintf("Hello user with UUID: %s", kws.UUID))) - kws.SetAttribute("user_id", kws.UUID) + kws.SetAttribute("is_tapper", false) })) - ikisocket.On(ikisocket.EventMessage, webSocketMessage) - ikisocket.On(ikisocket.EventConnect, webSocketConnect) - ikisocket.On(ikisocket.EventDisconnect, webSocketDisconnect) - ikisocket.On(ikisocket.EventClose, webSocketClose) // This event is called when the server disconnects the user actively with .Close() method - ikisocket.On(ikisocket.EventError, webSocketError) // On error event + app.Get("/wsTapper", ikisocket.New(func(kws *ikisocket.Websocket) { + // Tapper clients are handled differently, they don't need to receive new message broadcasts. + kws.SetAttribute("is_tapper", true) + })) + + ikisocket.On(ikisocket.EventMessage, eventHandlers.WebSocketMessage) + ikisocket.On(ikisocket.EventConnect, eventHandlers.WebSocketConnect) + ikisocket.On(ikisocket.EventDisconnect, eventHandlers.WebSocketDisconnect) + ikisocket.On(ikisocket.EventClose, eventHandlers.WebSocketClose) // This event is called when the server disconnects the user actively with .Close() method + ikisocket.On(ikisocket.EventError, eventHandlers.WebSocketError) // On error event } diff --git a/api/pkg/tap/passive_tapper.go b/api/pkg/tap/passive_tapper.go index 7858d2e94..b5cacc9e8 100644 --- a/api/pkg/tap/passive_tapper.go +++ b/api/pkg/tap/passive_tapper.go @@ -134,7 +134,7 @@ var nErrors uint var appPorts []int // global var ownIps []string //global var hostMode bool //global -var hostAppAddresses []string //global +var HostAppAddresses []string //global /* minOutputLevel: Error will be printed only if outputLevel is above this value * t: key for errorsMap (counting errors) @@ -198,7 +198,7 @@ func (c *Context) GetCaptureInfo() gopacket.CaptureInfo { return c.CaptureInfo } -func StartPassiveTapper() chan *OutputChannelItem { +func StartPassiveTapper() <-chan *OutputChannelItem { var harWriter *HarWriter if *dumpToHar { harWriter = NewHarWriter(*HarOutputDir, *harEntriesPerFile) @@ -240,8 +240,6 @@ func startPassiveTapper(harWriter *HarWriter) { } else { appPorts = parseAppPorts(appPortsStr) } - hostAppAddresses = parseHostAppAddresses(*hostAppAddressesString) - fmt.Println("Filtering for the following addresses:", hostAppAddresses) tapOutputPort := os.Getenv(OutPortEnvVar) if tapOutputPort == "" { fmt.Println("Received empty/no WEB_SOCKET_PORT env var! falling back to port 8080") @@ -275,7 +273,8 @@ func startPassiveTapper(harWriter *HarWriter) { appPorts = *parsedMessage.Ports } else if parsedMessage.MessageType == "setAddresses" { Debug("Got message from collector. Type: %s, IPs: %v\n", parsedMessage.MessageType, parsedMessage.Addresses) - hostAppAddresses = *parsedMessage.Addresses + HostAppAddresses = *parsedMessage.Addresses + Info("Filtering for the following addresses: %s\n", HostAppAddresses) } } else { Error("Collector-Message-Parsing", "Error parsing message from collector: %s (%v,%+v)\n", err, err, err) diff --git a/api/pkg/tap/tcp_stream_factory.go b/api/pkg/tap/tcp_stream_factory.go index 3cb7bfd2d..23bda51bd 100644 --- a/api/pkg/tap/tcp_stream_factory.go +++ b/api/pkg/tap/tcp_stream_factory.go @@ -85,9 +85,9 @@ func (factory *tcpStreamFactory) WaitGoRoutines() { func (factory *tcpStreamFactory) shouldTap(dstIP string, dstPort int) bool { if hostMode { - if inArrayString(hostAppAddresses, fmt.Sprintf("%s:%d", dstIP, dstPort)) == true { + if inArrayString(HostAppAddresses, fmt.Sprintf("%s:%d", dstIP, dstPort)) == true { return true - } else if inArrayString(hostAppAddresses, dstIP) == true { + } else if inArrayString(HostAppAddresses, dstIP) == true { return true } return false diff --git a/cli/cmd/tap.go b/cli/cmd/tap.go index 6985d9422..4b8766bcb 100644 --- a/cli/cmd/tap.go +++ b/cli/cmd/tap.go @@ -2,6 +2,8 @@ package cmd import ( "errors" + "fmt" + "regexp" "github.com/spf13/cobra" @@ -10,20 +12,25 @@ import ( ) var tapCmd = &cobra.Command{ - Use: "tap [PODNAME]", + Use: "tap [POD REGEX]", Short: "Record ingoing traffic of a kubernetes pod", Long: `Record the ingoing traffic of a kubernetes pod. Supported protocols are HTTP and gRPC.`, RunE: func(cmd *cobra.Command, args []string) error { if len(args) == 0 { - return errors.New("PODNAME argument is required") + return errors.New("POD REGEX argument is required") } else if len(args) > 1 { return errors.New("Unexpected number of arguments") } - podName := args[0] + regex, err := regexp.Compile(args[0]) + if err != nil { - mizu.Run(podName) + return errors.New(fmt.Sprintf("%s is not a valid regex %s", args[0], err)) + return nil + } + + mizu.Run(regex) return nil }, } diff --git a/cli/kubernetes/provider.go b/cli/kubernetes/provider.go index b0c9f3ae3..62b39a1d5 100644 --- a/cli/kubernetes/provider.go +++ b/cli/kubernetes/provider.go @@ -3,12 +3,15 @@ package kubernetes import ( _ "bytes" "context" + "encoding/json" "errors" "fmt" + apps "k8s.io/api/apps/v1" core "k8s.io/api/core/v1" rbac "k8s.io/api/rbac/v1" k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/kubernetes" _ "k8s.io/client-go/plugin/pkg/client/auth/azure" @@ -20,7 +23,7 @@ import ( _ "k8s.io/client-go/tools/portforward" "k8s.io/client-go/util/homedir" "path/filepath" - "strings" + "regexp" ) type Provider struct { @@ -77,48 +80,30 @@ func (provider *Provider) GetPods(ctx context.Context, namespace string) { fmt.Printf("There are %d pods in Namespace %s\n", len(pods.Items), namespace) } -func (provider *Provider) CreateMizuPod(ctx context.Context, namespace string, podName string, podImage string, tappedPodNamespace string, tappedPodName string, linkServiceAccount bool) (*core.Pod, error) { - tappedPod, err := provider.clientSet.CoreV1().Pods(tappedPodNamespace).Get(ctx, tappedPodName, metav1.GetOptions{}) - if err != nil { - panic(err.Error()) - } - - podIps := make([]string, len(tappedPod.Status.PodIPs)) - for ii, podIp := range tappedPod.Status.PodIPs { - podIps[ii] = podIp.IP - } - podIpsString := strings.Join(podIps, ",") - - privileged := true +func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace string, podName string, podImage string, linkServiceAccount bool) (*core.Pod, error) { pod := &core.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: podName, Namespace: namespace, + Labels: map[string]string{"app": podName}, }, Spec: core.PodSpec{ - HostNetwork: true, // very important to make passive tapper see traffic Containers: []core.Container{ { Name: podName, Image: podImage, ImagePullPolicy: core.PullAlways, - SecurityContext: &core.SecurityContext{ - Privileged: &privileged, // must be privileged to get node level traffic - }, + Command: []string {"./mizuagent", "--aggregator"}, Env: []core.EnvVar{ { Name: "HOST_MODE", Value: "1", }, - { - Name: "TAPPED_ADDRESSES", - Value: podIpsString, - }, }, }, }, + DNSPolicy: "ClusterFirstWithHostNet", TerminationGracePeriodSeconds: new(int64), - NodeSelector: map[string]string{"kubernetes.io/hostname": tappedPod.Spec.NodeName}, }, } //define the service account only when it exists to prevent pod crash @@ -128,6 +113,21 @@ func (provider *Provider) CreateMizuPod(ctx context.Context, namespace string, p return provider.clientSet.CoreV1().Pods(namespace).Create(ctx, pod, metav1.CreateOptions{}) } +func (provider *Provider) CreateService(ctx context.Context, namespace string, serviceName string, appLabelValue string) (*core.Service, error) { + service := core.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: serviceName, + Namespace: namespace, + }, + Spec: core.ServiceSpec{ + Ports: []core.ServicePort {{TargetPort: intstr.FromInt(8899), Port: 80}}, + Type: core.ServiceTypeClusterIP, + Selector: map[string]string{"app": appLabelValue}, + }, + } + return provider.clientSet.CoreV1().Services(namespace).Create(ctx, &service, metav1.CreateOptions{}) +} + func (provider *Provider) DoesMizuRBACExist(ctx context.Context, namespace string) (bool, error){ serviceAccount, err := provider.clientSet.CoreV1().ServiceAccounts(namespace).Get(ctx, serviceAccountName, metav1.GetOptions{}) @@ -200,8 +200,102 @@ func (provider *Provider) CreateMizuRBAC(ctx context.Context, namespace string , return nil } -func (provider *Provider) RemovePod(ctx context.Context, namespace string, podName string) { - provider.clientSet.CoreV1().Pods(namespace).Delete(ctx, podName, metav1.DeleteOptions{}) +func (provider *Provider) RemovePod(ctx context.Context, namespace string, podName string) error { + return provider.clientSet.CoreV1().Pods(namespace).Delete(ctx, podName, metav1.DeleteOptions{}) +} + +func (provider *Provider) RemoveService(ctx context.Context, namespace string, serviceName string) error { + return provider.clientSet.CoreV1().Services(namespace).Delete(ctx, serviceName, metav1.DeleteOptions{}) +} + +func (provider *Provider) RemoveDaemonSet(ctx context.Context, namespace string, daemonSetName string) error { + return provider.clientSet.AppsV1().DaemonSets(namespace).Delete(ctx, daemonSetName, metav1.DeleteOptions{}) +} + +func (provider *Provider) CreateMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, aggregatorPodIp string, nodeToTappedPodIPMap map[string][]string, linkServiceAccount bool) error { + nodeToTappedPodIPMapJsonStr, err := json.Marshal(nodeToTappedPodIPMap) + if err != nil { + return err + } + + privileged := true + podTemplate := core.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{"app": tapperPodName}, + }, + Spec: core.PodSpec{ + HostNetwork: true, // very important to make passive tapper see traffic + Containers: []core.Container{ + { + Name: tapperPodName, + Image: podImage, + ImagePullPolicy: core.PullAlways, + SecurityContext: &core.SecurityContext{ + Privileged: &privileged, // must be privileged to get node level traffic + }, + Command: []string {"./mizuagent", "-i", "any", "--tap", "--hardump", "--aggregator-address", fmt.Sprintf("ws://%s/wsTapper", aggregatorPodIp)}, + Env: []core.EnvVar{ + { + Name: "HOST_MODE", + Value: "1", + }, + { + Name: "AGGREGATOR_ADDRESS", + Value: aggregatorPodIp, + }, + { + Name: "TAPPED_ADDRESSES_PER_HOST", + Value: string(nodeToTappedPodIPMapJsonStr), + }, + { + Name: "NODE_NAME", + ValueFrom: &core.EnvVarSource{ + FieldRef: &core.ObjectFieldSelector { + APIVersion: "v1", + FieldPath: "spec.nodeName", + }, + }, + }, + }, + }, + }, + DNSPolicy: "ClusterFirstWithHostNet", + TerminationGracePeriodSeconds: new(int64), + // Affinity: TODO: define node selector for all relevant nodes for this mizu instance + }, + } + if linkServiceAccount { + podTemplate.Spec.ServiceAccountName = serviceAccountName + } + labelSelector := metav1.LabelSelector{ + MatchLabels: map[string]string{"app": tapperPodName}, + } + daemonSet := apps.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: daemonSetName, + Namespace: namespace, + }, + Spec: apps.DaemonSetSpec{ + Selector: &labelSelector, + Template: podTemplate, + }, + } + _, err = provider.clientSet.AppsV1().DaemonSets(namespace).Create(ctx, &daemonSet, metav1.CreateOptions{}) + return err +} + +func (provider *Provider) GetAllPodsMatchingRegex(ctx context.Context, regex *regexp.Regexp) ([]core.Pod, error) { + pods, err := provider.clientSet.CoreV1().Pods("").List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, err + } + matchingPods := make([]core.Pod, 0) + for _, pod := range pods.Items { + if regex.MatchString(pod.Name) { + matchingPods = append(matchingPods, pod) + } + } + return matchingPods, err } func getClientSet(config *restclient.Config) *kubernetes.Clientset { diff --git a/cli/mizu/consts.go b/cli/mizu/consts.go index 356da23ab..20a6db174 100644 --- a/cli/mizu/consts.go +++ b/cli/mizu/consts.go @@ -7,4 +7,7 @@ var ( const ( MizuResourcesNamespace = "default" + TapperDaemonSetName = "mizu-tapper-daemon-set" + aggregatorPodName = "mizu-collector" + tapperPodName = "mizu-tapper" ) diff --git a/cli/mizu/mizuRunner.go b/cli/mizu/mizuRunner.go index bbc6cb099..3d7525751 100644 --- a/cli/mizu/mizuRunner.go +++ b/cli/mizu/mizuRunner.go @@ -12,56 +12,90 @@ import ( "time" ) -func Run(tappedPodName string) { +func Run(podRegexQuery *regexp.Regexp) { kubernetesProvider := kubernetes.NewProvider(config.Configuration.KubeConfigPath, config.Configuration.Namespace) ctx, cancel := context.WithCancel(context.Background()) defer cancel() // cancel will be called when this function exits - podName := "mizu-collector" - - mizuServiceAccountExists := createRBACIfNecessary(ctx, kubernetesProvider) - go createPodAndPortForward(ctx, kubernetesProvider, cancel, podName, MizuResourcesNamespace, tappedPodName, mizuServiceAccountExists) //TODO convert this to job for built in pod ttl or have the running app handle this - waitForFinish(ctx, cancel) //block until exit signal or error + nodeToTappedPodIPMap, err := getNodeHostToTappedPodIpsMap(ctx, kubernetesProvider, podRegexQuery) + if err != nil { + cleanUpMizuResources(kubernetesProvider) + return + } + err = createMizuResources(ctx, kubernetesProvider, nodeToTappedPodIPMap) + if err != nil { + cleanUpMizuResources(kubernetesProvider) + return + } + go portForwardApiPod(ctx, kubernetesProvider, cancel) //TODO convert this to job for built in pod ttl or have the running app handle this + waitForFinish(ctx, cancel) //block until exit signal or error // TODO handle incoming traffic from tapper using a channel //cleanup - fmt.Printf("\nremoving pod %s\n", podName) - removalCtx, _ := context.WithTimeout(context.Background(), 2 * time.Second) - kubernetesProvider.RemovePod(removalCtx, MizuResourcesNamespace, podName) + fmt.Printf("\nRemoving mizu resources\n") + cleanUpMizuResources(kubernetesProvider) } -func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, podRegex *regexp.Regexp) { - added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx, kubernetesProvider.Namespace), podRegex) - for { - select { - case newTarget := <- added: - fmt.Printf("+%s\n", newTarget.Name) - - case removedTarget := <- removed: - fmt.Printf("-%s\n", removedTarget.Name) - - case <- modified: - continue - - case <- errorChan: - cancel() - - case <- ctx.Done(): - return - } - } -} - -func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, podName string, namespace string, tappedPodName string, linkServiceAccount bool) { - pod, err := kubernetesProvider.CreateMizuPod(ctx, MizuResourcesNamespace, podName, config.Configuration.MizuImage, kubernetesProvider.Namespace, tappedPodName, linkServiceAccount) +func createMizuResources(ctx context.Context, kubernetesProvider *kubernetes.Provider, nodeToTappedPodIPMap map[string][]string) error { + mizuServiceAccountExists := createRBACIfNecessary(ctx, kubernetesProvider) + _, err := kubernetesProvider.CreateMizuAggregatorPod(ctx, MizuResourcesNamespace, aggregatorPodName, config.Configuration.MizuImage, mizuServiceAccountExists) if err != nil { - fmt.Printf("error creating pod %s", err) - cancel() - return + fmt.Printf("Error creating mizu collector pod: %v\n", err) + return err } - podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s$", pod.Name)) - added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx, namespace), podExactRegex) + aggregatorService, err := kubernetesProvider.CreateService(ctx, MizuResourcesNamespace, aggregatorPodName, aggregatorPodName) + if err != nil { + fmt.Printf("Error creating mizu collector service: %v\n", err) + return err + } + err = kubernetesProvider.CreateMizuTapperDaemonSet(ctx, MizuResourcesNamespace, TapperDaemonSetName, config.Configuration.MizuImage, tapperPodName, fmt.Sprintf("%s.%s.svc.cluster.local", aggregatorService.Name, aggregatorService.Namespace), nodeToTappedPodIPMap, mizuServiceAccountExists) + if err != nil { + fmt.Printf("Error creating mizu tapper daemonset: %v\n", err) + return err + } + return nil +} + +func cleanUpMizuResources(kubernetesProvider *kubernetes.Provider) { + removalCtx, _ := context.WithTimeout(context.Background(), 5 * time.Second) + if err := kubernetesProvider.RemovePod(removalCtx, MizuResourcesNamespace, aggregatorPodName); err != nil { + fmt.Printf("Error removing Pod %s in namespace %s: %s (%v,%+v)\n", aggregatorPodName, MizuResourcesNamespace, err, err, err); + } + if err := kubernetesProvider.RemoveService(removalCtx, MizuResourcesNamespace, aggregatorPodName); err != nil { + fmt.Printf("Error removing Service %s in namespace %s: %s (%v,%+v)\n", aggregatorPodName, MizuResourcesNamespace, err, err, err); + } + if err := kubernetesProvider.RemoveDaemonSet(removalCtx, MizuResourcesNamespace, TapperDaemonSetName); err != nil { + fmt.Printf("Error removing DaemonSet %s in namespace %s: %s (%v,%+v)\n", TapperDaemonSetName, MizuResourcesNamespace, err, err, err); + } +} + +// will be relevant in the future +//func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, podRegex *regexp.Regexp) { +// added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx, kubernetesProvider.Namespace), podRegex) +// for { +// select { +// case newTarget := <- added: +// fmt.Printf("+%s\n", newTarget.Name) +// +// case removedTarget := <- removed: +// fmt.Printf("-%s\n", removedTarget.Name) +// +// case <- modified: +// continue +// +// case <- errorChan: +// cancel() +// +// case <- ctx.Done(): +// return +// } +// } +//} + +func portForwardApiPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) { + podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s$", aggregatorPodName)) + added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx, MizuResourcesNamespace), podExactRegex) isPodReady := false var portForward *kubernetes.PortForward for { @@ -69,14 +103,14 @@ func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes case <- added: continue case <- removed: - fmt.Printf("%s removed\n", podName) + fmt.Printf("%s removed\n", aggregatorPodName) cancel() return case modifiedPod := <- modified: if modifiedPod.Status.Phase == "Running" && !isPodReady { isPodReady = true var err error - portForward, err = kubernetes.NewPortForward(kubernetesProvider, namespace, podName, config.Configuration.GuiPort, config.Configuration.MizuPodPort, cancel) + portForward, err = kubernetes.NewPortForward(kubernetesProvider, MizuResourcesNamespace, aggregatorPodName, config.Configuration.GuiPort, config.Configuration.MizuPodPort, cancel) fmt.Printf("Web interface is now available at http://localhost:%d\n", config.Configuration.GuiPort) if err != nil { fmt.Printf("error forwarding port to pod %s\n", err) @@ -86,7 +120,7 @@ func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes case <- time.After(25 * time.Second): if !isPodReady { - fmt.Printf("error: %s pod was not ready in time", podName) + fmt.Printf("error: %s pod was not ready in time", aggregatorPodName) cancel() } @@ -122,6 +156,23 @@ func createRBACIfNecessary(ctx context.Context, kubernetesProvider *kubernetes.P return true } +func getNodeHostToTappedPodIpsMap(ctx context.Context, kubernetesProvider *kubernetes.Provider, regex *regexp.Regexp) (map[string][]string, error) { + matchingPods, err := kubernetesProvider.GetAllPodsMatchingRegex(ctx, regex) + if err != nil { + return nil, err + } + nodeToTappedPodIPMap := make(map[string][]string, 0) + for _, pod := range matchingPods { + existingList := nodeToTappedPodIPMap[pod.Spec.NodeName] + if existingList == nil { + nodeToTappedPodIPMap[pod.Spec.NodeName] = []string {pod.Status.PodIP} + } else { + nodeToTappedPodIPMap[pod.Spec.NodeName] = append(nodeToTappedPodIPMap[pod.Spec.NodeName], pod.Status.PodIP) + } + } + return nodeToTappedPodIPMap, nil +} + func waitForFinish(ctx context.Context, cancel context.CancelFunc) { sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)