pubsub fix and refactor

Author: subhashish-devtron

App.go

@@ -39,7 +39,7 @@ func NewApp(Router *api.Router, Logger *zap.SugaredLogger,
 }
 
 type ServerConfig struct {
-	SERVER_HTTP_PORT int `env:"SERVER_HTTP_PORT" envDefault:"8080"`
+	SERVER_HTTP_PORT int `env:"SERVER_HTTP_PORT" envDefault:"8083"`
 }
 
 func (app *App) Start() {

Wire.go

@@ -37,8 +37,8 @@ func InitializeApp() (*App, error) {
 		pubsub.NewNatSubscription,
 		grafeasService.NewKlarServiceImpl,
 		wire.Bind(new(grafeasService.GrafeasService), new(*grafeasService.GrafeasServiceImpl)),
-		pubsub.NewTestPublishImpl,
-		wire.Bind(new(pubsub.TestPublish), new(*pubsub.TestPublishImpl)),
+		//pubsub.NewTestPublishImpl,
+		//wire.Bind(new(pubsub.TestPublish), new(*pubsub.TestPublishImpl)),
 
 		clairService.GetClairConfig,
 		clairService.NewClairServiceImpl,

api/RestHandler.go

@@ -9,26 +9,27 @@ import (
 	"github.com/devtron-labs/image-scanner/pkg/klarService"
 	"github.com/devtron-labs/image-scanner/pkg/security"
 	"github.com/devtron-labs/image-scanner/pkg/user"
-	"github.com/devtron-labs/image-scanner/pubsub"
+	//"github.com/devtron-labs/image-scanner/pubsub"
 	"go.uber.org/zap"
 	"net/http"
 	"os"
 )
 
 type RestHandler interface {
 	ScanForVulnerability(w http.ResponseWriter, r *http.Request)
+	ScanForVulnerabilityEvent(scanConfig *common.ImageScanEvent) (*common.ScanEventResponse, error)
 }
 
 func NewRestHandlerImpl(logger *zap.SugaredLogger,
-	testPublish pubsub.TestPublish,
+	//testPublish pubsub.TestPublish,
 	grafeasService grafeasService.GrafeasService,
 	userService user.UserService, imageScanService security.ImageScanService,
 	klarService klarService.KlarService,
 	clairService clairService.ClairService,
 	imageScanConfig *security.ImageScanConfig) *RestHandlerImpl {
 	return &RestHandlerImpl{
-		logger:           logger,
-		testPublish:      testPublish,
+		logger: logger,
+		//testPublish:      testPublish,
 		grafeasService:   grafeasService,
 		userService:      userService,
 		imageScanService: imageScanService,
@@ -39,8 +40,8 @@ func NewRestHandlerImpl(logger *zap.SugaredLogger,
 }
 
 type RestHandlerImpl struct {
-	logger           *zap.SugaredLogger
-	testPublish      pubsub.TestPublish
+	logger *zap.SugaredLogger
+	//testPublish      pubsub.TestPublish
 	grafeasService   grafeasService.GrafeasService
 	userService      user.UserService
 	imageScanService security.ImageScanService
@@ -76,6 +77,17 @@ func (impl *RestHandlerImpl) ScanForVulnerability(w http.ResponseWriter, r *http
 		writeJsonResp(w, err, nil, http.StatusBadRequest)
 		return
 	}
+	result, err := impl.ScanForVulnerabilityEvent(&scanConfig)
+	if err != nil {
+		writeJsonResp(w, err, nil, http.StatusInternalServerError)
+		return
+	}
+	impl.logger.Debugw("save", "status", result)
+	writeJsonResp(w, err, result, http.StatusOK)
+}
+
+func (impl *RestHandlerImpl) ScanForVulnerabilityEvent(scanConfig *common.ImageScanEvent) (*common.ScanEventResponse, error) {
+
 	if scanConfig.UserId == 0 {
 		scanConfig.UserId = 1 //setting user as system user in case of empty user data
 	}
@@ -84,45 +96,37 @@ func (impl *RestHandlerImpl) ScanForVulnerability(w http.ResponseWriter, r *http
 	tool, err := impl.imageScanService.GetActiveTool()
 	if err != nil {
 		impl.logger.Errorw("err in image scanning", "err", err)
-		writeJsonResp(w, err, nil, http.StatusInternalServerError)
-		return
+		return nil, err
 	}
-	executionHistory, executionHistoryDirPath, err := impl.imageScanService.RegisterScanExecutionHistoryAndState(&scanConfig, tool)
+	executionHistory, executionHistoryDirPath, err := impl.imageScanService.RegisterScanExecutionHistoryAndState(scanConfig, tool)
 	if err != nil {
 		impl.logger.Errorw("service err, RegisterScanExecutionHistoryAndState", "err", err)
-		writeJsonResp(w, err, nil, http.StatusInternalServerError)
-		return
+		return nil, err
 	}
 	if tool.Name == bean.ScanToolClair && tool.Version == bean.ScanToolVersion2 {
-		result, err = impl.klarService.Process(&scanConfig, executionHistory)
+		result, err = impl.klarService.Process(scanConfig, executionHistory)
 		if err != nil {
 			impl.logger.Errorw("err in process msg", "err", err)
-			writeJsonResp(w, err, nil, http.StatusInternalServerError)
-			return
+			return nil, err
 		}
 	} else if tool.Name == bean.ScanToolClair && tool.Version == bean.ScanToolVersion4 {
-		result, err = impl.clairService.ScanImage(&scanConfig, tool, executionHistory)
+		result, err = impl.clairService.ScanImage(scanConfig, tool, executionHistory)
 		if err != nil {
 			impl.logger.Errorw("err in process msg", "err", err)
-			writeJsonResp(w, err, nil, http.StatusInternalServerError)
-			return
+			return nil, err
 		}
 	} else {
-		err = impl.imageScanService.ScanImage(&scanConfig, tool, executionHistory, executionHistoryDirPath)
+		err = impl.imageScanService.ScanImage(scanConfig, tool, executionHistory, executionHistoryDirPath)
 		if err != nil {
 			impl.logger.Errorw("err in process msg", "err", err)
-			writeJsonResp(w, err, nil, http.StatusInternalServerError)
-			return
+			return nil, err
 		}
 	}
 	//deleting executionDirectoryPath with files as well
 	err = os.RemoveAll(executionHistoryDirPath)
 	if err != nil {
 		impl.logger.Errorw("error in deleting executionHistoryDirectory", "err", err)
-		writeJsonResp(w, err, nil, http.StatusInternalServerError)
-		return
+		return nil, err
 	}
-
-	impl.logger.Debugw("save", "status", result)
-	writeJsonResp(w, err, result, http.StatusOK)
+	return result, nil
 }

internal/sql/bean/bean.go

@@ -7,7 +7,7 @@ import (
 )
 
 const (
-	ScanOutputDirectory      = "/security/devtronimagescan" // This is not configurable due to permissions for devtron-user
+	ScanOutputDirectory      = "/Users/subhashish/tmp/security/devtronimagescan" // This is not configurable due to permissions for devtron-user
 	NullProcessIndex         = -1
 	JsonOutputFileNameSuffix = "_out.json"
 	ScannerTypeClairV4       = "CLAIRV4"

pubsub/NatSubscription.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	pubsub1 "github.com/devtron-labs/common-lib/pubsub-lib"
 	"github.com/devtron-labs/common-lib/pubsub-lib/model"
+	"github.com/devtron-labs/image-scanner/api"
 	"github.com/devtron-labs/image-scanner/common"
 	"github.com/devtron-labs/image-scanner/pkg/clairService"
 	"go.uber.org/zap"
@@ -17,15 +18,18 @@ type NatSubscriptionImpl struct {
 	pubSubClient *pubsub1.PubSubClientServiceImpl
 	logger       *zap.SugaredLogger
 	clairService clairService.ClairService
+	restHandler  *api.RestHandlerImpl
 }
 
 func NewNatSubscription(pubSubClient *pubsub1.PubSubClientServiceImpl,
 	logger *zap.SugaredLogger,
-	clairService clairService.ClairService) (*NatSubscriptionImpl, error) {
+	clairService clairService.ClairService,
+	restHandler *api.RestHandlerImpl) (*NatSubscriptionImpl, error) {
 	ns := &NatSubscriptionImpl{
 		pubSubClient: pubSubClient,
 		logger:       logger,
 		clairService: clairService,
+		restHandler:  restHandler,
 	}
 	return ns, ns.Subscribe()
 }
@@ -44,7 +48,8 @@ func (impl *NatSubscriptionImpl) Subscribe() error {
 		// NOTE: This is not being used, thats why not updated the call
 		// TODO: Will have to update if any usage in future
 		// scanConfig.Image = "quay.io/coreos/clair:v2.0.0"
-		_, err = impl.clairService.ScanImage(scanConfig, nil, nil)
+		_, err = impl.restHandler.ScanForVulnerabilityEvent(scanConfig)
+		//_, err = impl.clairService.ScanImage(scanConfig, nil, nil)
 		if err != nil {
 			impl.logger.Infow("err in process msg", "err", err)
 			return