Move alerts and workflows API to graffiti

.mk/api.mk

@@ -1,5 +1,8 @@
 .typescript: statics/js/bundle.js js/browser.js js/api.js
 
+graffiti/js/api.js: graffiti/js/api.ts
+	cd graffiti/js && npm ci && PATH=`npm bin`:$$PATH tsc --module commonjs --target ES5 api.ts
+
 js/api.js: js/api.ts
 	cd js && npm ci && PATH=`npm bin`:$$PATH tsc --module commonjs --target ES5 api.ts
 

.mk/bindata.mk

@@ -28,3 +28,7 @@ ebpf/statics/bindata.go: $(EBPF_PROBES)
 statics/bindata.go: statics/js/bundle.js $(STATIC_FILES)
 	go run ${GO_BINDATA_GITHUB} ${GO_BINDATA_FLAGS} -nometadata -o statics/bindata.go -pkg=statics -ignore=bindata.go $(BINDATA_DIRS)
 	gofmt -w -s statics/bindata.go
+
+graffiti/js/bindata.go: graffiti/js/*.js
+	go run ${GO_BINDATA_GITHUB} ${GO_BINDATA_FLAGS} -prefix graffiti/js/ -nometadata -o graffiti/js/bindata.go -pkg=js graffiti/js/*.js
+	gofmt -w -s graffiti/js/bindata.go

analyzer/server.go

@@ -21,13 +21,14 @@ package analyzer
 
 import (
 	"crypto/tls"
-	"errors"
 	"fmt"
 	"net/http"
 	"time"
 
+	"github.com/pkg/errors"
+	yaml "gopkg.in/yaml.v2"
+
 	"github.com/skydive-project/dede/dede"
-	"github.com/skydive-project/skydive/alert"
 	api "github.com/skydive-project/skydive/api/server"
 	"github.com/skydive-project/skydive/api/types"
 	"github.com/skydive-project/skydive/config"
@@ -37,6 +38,7 @@ import (
 	"github.com/skydive-project/skydive/flow/storage"
 	"github.com/skydive-project/skydive/graffiti/api/rest"
 	gapi "github.com/skydive-project/skydive/graffiti/api/server"
+	gtypes "github.com/skydive-project/skydive/graffiti/api/types"
 	etcdclient "github.com/skydive-project/skydive/graffiti/etcd/client"
 	etcdserver "github.com/skydive-project/skydive/graffiti/etcd/server"
 	"github.com/skydive-project/skydive/graffiti/graph"
@@ -49,19 +51,15 @@ import (
 	"github.com/skydive-project/skydive/packetinjector"
 	"github.com/skydive-project/skydive/probe"
 	"github.com/skydive-project/skydive/sflow"
+	"github.com/skydive-project/skydive/statics"
 	"github.com/skydive-project/skydive/topology"
 	usertopology "github.com/skydive-project/skydive/topology/enhancers"
 	"github.com/skydive-project/skydive/topology/probes/blockdev"
 	"github.com/skydive-project/skydive/ui"
 	"github.com/skydive-project/skydive/validator"
 )
 
-// ElectionStatus describes the status of an election
-//
-// easyjson:json
-type ElectionStatus struct {
-	IsMaster bool
-}
+const workflowAssetDir = "statics/workflows"
 
 // Status analyzer object
 //
@@ -74,16 +72,15 @@ type Status struct {
 	Peers       hub.PeersStatus
 	Publishers  map[string]ws.ConnStatus
 	Subscribers map[string]ws.ConnStatus
-	Alerts      ElectionStatus
-	Captures    ElectionStatus
+	Alerts      hub.ElectionStatus
+	Captures    hub.ElectionStatus
 	Probes      map[string]interface{}
 }
 
 // Server describes an Analyzer servers mechanism like http, websocket, topology, ondemand probes, ...
 type Server struct {
 	uiServer        *ui.Server
 	hub             *hub.Hub
-	alertServer     *alert.Server
 	onDemandClient  *client.OnDemandClient
 	piClient        *client.OnDemandClient
 	topologyManager *usertopology.TopologyManager
@@ -103,8 +100,8 @@ func (s *Server) GetStatus() interface{} {
 		Peers:       hubStatus.Peers,
 		Publishers:  hubStatus.Publishers,
 		Subscribers: hubStatus.Subscribers,
-		Alerts:      ElectionStatus{IsMaster: s.alertServer.IsMaster()},
-		Captures:    ElectionStatus{IsMaster: s.onDemandClient.IsMaster()},
+		Alerts:      hubStatus.Alerts,
+		Captures:    hub.ElectionStatus{IsMaster: s.onDemandClient.IsMaster()},
 		Probes:      s.probeBundle.GetStatus(),
 	}
 }
@@ -129,6 +126,27 @@ func (s *Server) createStartupCapture() error {
 	return nil
 }
 
+func (s *Server) loadStaticWorkflows() error {
+	assets, err := statics.AssetDir(workflowAssetDir)
+	if err == nil {
+		for _, asset := range assets {
+			yml, err := statics.Asset(workflowAssetDir + "/" + asset)
+			if err != nil {
+				return err
+			}
+
+			var workflow gtypes.Workflow
+			if err := yaml.Unmarshal([]byte(yml), &workflow); err != nil {
+				return errors.Wrapf(err, "failed to load workflow %s", asset)
+			}
+
+			gapi.StaticWorkflows = append(gapi.StaticWorkflows, &workflow)
+		}
+	}
+
+	return nil
+}
+
 // Start the analyzer server
 func (s *Server) Start() error {
 	if err := s.hub.Start(); err != nil {
@@ -147,7 +165,6 @@ func (s *Server) Start() error {
 
 	s.onDemandClient.Start()
 	s.piClient.Start()
-	s.alertServer.Start()
 	s.topologyManager.Start()
 	s.flowServer.Start()
 
@@ -165,7 +182,6 @@ func (s *Server) Stop() {
 	s.probeBundle.Stop()
 	s.onDemandClient.Stop()
 	s.piClient.Stop()
-	s.alertServer.Stop()
 	s.topologyManager.Stop()
 	s.etcdClient.Stop()
 
@@ -322,64 +338,29 @@ func NewServerFromConfig() (*Server, error) {
 
 	apiServer := hub.APIServer()
 
-	captureAPIHandler, err := api.RegisterCaptureAPI(apiServer, g, apiAuthBackend)
-	if err != nil {
-		return nil, err
-	}
-
-	piAPIHandler, err := api.RegisterPacketInjectorAPI(g, apiServer, apiAuthBackend)
-	if err != nil {
-		return nil, err
-	}
+	captureAPIHandler := api.RegisterCaptureAPI(apiServer, g, apiAuthBackend)
 
+	piAPIHandler := api.RegisterPacketInjectorAPI(g, apiServer, apiAuthBackend)
 	s.piClient = packetinjector.NewOnDemandInjectionClient(g, piAPIHandler, hub.PodServer(), hub.SubscriberServer(), etcdClient)
 
-	_, err = gapi.RegisterNodeAPI(apiServer, g, apiAuthBackend)
-	if err != nil {
-		return nil, err
-	}
-
-	nodeRuleAPIHandler, err := api.RegisterNodeRuleAPI(apiServer, g, apiAuthBackend)
-	if err != nil {
-		return nil, err
-	}
-
-	_, err = gapi.RegisterEdgeAPI(apiServer, g, apiAuthBackend)
-	if err != nil {
-		return nil, err
-	}
-
-	edgeRuleAPIHandler, err := api.RegisterEdgeRuleAPI(apiServer, g, apiAuthBackend)
-	if err != nil {
-		return nil, err
-	}
-
+	nodeRuleAPIHandler := api.RegisterNodeRuleAPI(apiServer, g, apiAuthBackend)
+	edgeRuleAPIHandler := api.RegisterEdgeRuleAPI(apiServer, g, apiAuthBackend)
 	s.topologyManager = usertopology.NewTopologyManager(etcdClient, nodeRuleAPIHandler, edgeRuleAPIHandler, g)
 
-	if _, err = api.RegisterAlertAPI(apiServer, apiAuthBackend); err != nil {
-		return nil, err
-	}
-
-	if _, err := api.RegisterWorkflowAPI(apiServer, apiAuthBackend); err != nil {
-		return nil, err
-	}
-
 	s.onDemandClient = ondemand.NewOnDemandFlowProbeClient(g, captureAPIHandler, hub.PodServer(), hub.SubscriberServer(), etcdClient)
 
 	s.flowServer, err = server.NewFlowServer(hub.HTTPServer(), g, s.flowStorage, flowSubscriberEndpoint, probeBundle, clusterAuthBackend)
 	if err != nil {
 		return nil, err
 	}
 
-	s.alertServer, err = alert.NewServer(apiServer, hub.SubscriberServer(), g, tr, etcdClient)
-	if err != nil {
-		return nil, err
-	}
-
 	httpServer := hub.HTTPServer()
 	api.RegisterPcapAPI(httpServer, s.flowStorage, apiAuthBackend)
 	api.RegisterConfigAPI(httpServer, apiAuthBackend)
-	api.RegisterWorkflowCallAPI(httpServer, apiAuthBackend, apiServer, g, tr)
+
+	if err := s.loadStaticWorkflows(); err != nil {
+		return nil, err
+	}
 
 	if config.GetBool("analyzer.ssh_enabled") {
 		if err := dede.RegisterHandler("terminal", "/dede", httpServer.Router); err != nil {

api/server/capture.go

@@ -151,16 +151,14 @@ func (c *CaptureAPIHandler) Update(id string, resource rest.Resource) (rest.Reso
 }
 
 // RegisterCaptureAPI registers an new resource, capture
-func RegisterCaptureAPI(apiServer *api.Server, g *graph.Graph, authBackend shttp.AuthenticationBackend) (*CaptureAPIHandler, error) {
+func RegisterCaptureAPI(apiServer *api.Server, g *graph.Graph, authBackend shttp.AuthenticationBackend) *CaptureAPIHandler {
 	captureAPIHandler := &CaptureAPIHandler{
 		BasicAPIHandler: rest.BasicAPIHandler{
 			ResourceHandler: &CaptureResourceHandler{},
 			EtcdClient:      apiServer.EtcdClient,
 		},
 		Graph: g,
 	}
-	if err := apiServer.RegisterAPIHandler(captureAPIHandler, authBackend); err != nil {
-		return nil, err
-	}
-	return captureAPIHandler, nil
+	apiServer.RegisterAPIHandler(captureAPIHandler, authBackend)
+	return captureAPIHandler
 }

api/server/edgerule.go

@@ -55,17 +55,14 @@ func (a *EdgeRuleAPI) Update(id string, resource rest.Resource) (rest.Resource,
 }
 
 // RegisterEdgeRuleAPI registers an EdgeRule's API to a designated API Server
-func RegisterEdgeRuleAPI(apiServer *api.Server, g *graph.Graph, authBackend shttp.AuthenticationBackend) (*EdgeRuleAPI, error) {
+func RegisterEdgeRuleAPI(apiServer *api.Server, g *graph.Graph, authBackend shttp.AuthenticationBackend) *EdgeRuleAPI {
 	era := &EdgeRuleAPI{
 		BasicAPIHandler: rest.BasicAPIHandler{
 			ResourceHandler: &EdgeRuleResourceHandler{},
 			EtcdClient:      apiServer.EtcdClient,
 		},
 		Graph: g,
 	}
-	if err := apiServer.RegisterAPIHandler(era, authBackend); err != nil {
-		return nil, err
-	}
-
-	return era, nil
+	apiServer.RegisterAPIHandler(era, authBackend)
+	return era
 }