Merge pull request #105 from vijeyash1/obj-to-table

commit for issue  #94

agent/git/pkg/application/handlers.go

@@ -1,16 +1,19 @@
 package application
 
 import (
+	"encoding/json"
 	"log"
 	"net/http"
 
 	"github.com/gin-gonic/gin"
 	"github.com/intelops/kubviz/agent/git/api"
+	"github.com/intelops/kubviz/gitmodels/azuremodel"
+	"github.com/intelops/kubviz/model"
 )
 
 func (app *Application) PostGitea(c *gin.Context) {
-	repo := "Gitea"
-	event := c.Request.Header.Get("X-Gitea-Event")
+
+	event := c.Request.Header.Get(string(model.GiteaHeader))
 	if len(event) == 0 {
 		log.Println("error getting the gitea event from header")
 		return
@@ -19,69 +22,92 @@ func (app *Application) PostGitea(c *gin.Context) {
 	if err != nil {
 		log.Println("Error Reading Request Body")
 	}
-	app.conn.Publish(jsonData, repo, event)
+	app.conn.Publish(jsonData, string(model.GiteaProvider), model.GiteaHeader, model.EventValue(event))
 }
 
 func (app *Application) PostAzure(c *gin.Context) {
 
-	repo := "Azure"
 	jsonData, err := c.GetRawData()
 	if err != nil {
 		log.Println("Error Reading Request Body")
+		c.Status(http.StatusInternalServerError)
+		return
 	}
-	app.conn.Publish(jsonData, repo, "azure event")
+	var pl azuremodel.BasicEvent
+	err = json.Unmarshal([]byte(jsonData), &pl)
+	if err != nil {
+		log.Println("Error Reading Request Body")
+		c.Status(http.StatusInternalServerError)
+		return
+	}
+	event := pl.EventType
+	if string(event) == "" {
+		log.Println("Error Reading Request Body")
+		c.Status(http.StatusInternalServerError)
+		return
+	}
+	app.conn.Publish(jsonData, string(model.AzureDevopsProvider), model.AzureHeader, model.EventValue(event))
 }
 
 // githubHandler handles the github webhooks post requests.
 func (app *Application) PostGithub(c *gin.Context) {
-	repo := "Github"
-	event := c.Request.Header.Get("X-GitHub-Event")
+	event := c.Request.Header.Get(string(model.GithubHeader))
 	if len(event) == 0 {
 		log.Println("error getting the github event from header")
+		c.Status(http.StatusBadRequest)
 		return
 	}
 	jsonData, err := c.GetRawData()
 	if err != nil {
 		log.Println("Error Reading Request Body")
+		c.Status(http.StatusInternalServerError)
+		return
 	}
-	app.conn.Publish(jsonData, repo, event)
+	app.conn.Publish(jsonData, string(model.GithubProvider), model.GithubHeader, model.EventValue(event))
 }
 
 // gitlabHandler handles the github webhooks post requests.
 func (app *Application) PostGitlab(c *gin.Context) {
 
-	repo := "Gitlab"
-	event := c.Request.Header.Get("X-Gitlab-Event")
+	event := c.Request.Header.Get(string(model.GitlabHeader))
 	if len(event) == 0 {
 		log.Println("error getting the gitlab event from header")
+		c.Status(http.StatusBadRequest)
 		return
 	}
 	jsonData, err := c.GetRawData()
 	if err != nil {
 		log.Println("Error Reading Request Body")
+		c.Status(http.StatusInternalServerError)
+		return
 	}
-	app.conn.Publish(jsonData, repo, event)
+	app.conn.Publish(jsonData, string(model.GitlabProvider), model.GitlabHeader, model.EventValue(event))
 }
 
 // bitBucketHandler handles the github webhooks post requests.
 func (app *Application) PostBitbucket(c *gin.Context) {
 
-	repo := "BitBucket"
-	event := c.Request.Header.Get("X-Event-Key")
+	event := c.Request.Header.Get(string(model.BitBucketHeader))
 	if len(event) == 0 {
 		log.Println("error getting the bitbucket event from header")
+		c.Status(http.StatusBadRequest)
 		return
 	}
 	jsonData, err := c.GetRawData()
 	if err != nil {
 		log.Println("Error Reading Request Body")
+		c.Status(http.StatusInternalServerError)
+		return
 	}
-	app.conn.Publish(jsonData, repo, event)
+	app.conn.Publish(jsonData, string(model.BitBucketProvider), model.BitBucketHeader, model.EventValue(event))
 }
+
+// GetLiveness handles the liveness check get requests.
 func (app *Application) GetLiveness(c *gin.Context) {
 	c.Status(http.StatusOK)
 }
 
+// GetApiDocs handles the GetApiDocs get requests.
 func (app *Application) GetApiDocs(c *gin.Context) {
 	swagger, err := api.GetSwagger()
 	if err != nil {

agent/git/pkg/clients/nats_client.go

@@ -2,7 +2,10 @@ package clients
 
 import (
 	"fmt"
+
 	"github.com/intelops/kubviz/agent/git/pkg/config"
+	"github.com/intelops/kubviz/model"
+
 	"log"
 	"time"
 
@@ -87,11 +90,11 @@ func (n *NATSContext) Close() {
 	n.conn.Close()
 }
 
-func (n *NATSContext) Publish(metric []byte, repo, event string) error {
+func (n *NATSContext) Publish(metric []byte, repo string, eventkey model.EventKey, eventvalue model.EventValue) error {
 	msg := nats.NewMsg(eventSubject)
 	msg.Data = metric
-	msg.Header.Set("repo", repo)
-	msg.Header.Set("event", event)
+	msg.Header.Set("GitProvider", repo)
+	msg.Header.Set(string(eventkey), string(eventvalue))
 	_, err := n.stream.PublishMsgAsync(msg)
 
 	return err

client/pkg/clickhouse/db_client.go

@@ -5,13 +5,15 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
-	"github.com/ClickHouse/clickhouse-go/v2"
 	"log"
 	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2"
+
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 
 	"github.com/intelops/kubviz/client/pkg/config"
+	"github.com/intelops/kubviz/gitmodels/dbstatement"
 	"github.com/intelops/kubviz/model"
 )
 
@@ -35,6 +37,7 @@ type DBInterface interface {
 	RetrieveKubvizEvent() ([]model.DbEvent, error)
 	InsertContainerEventDockerHub(model.DockerHubBuild)
 	InsertContainerEventGithub(string)
+	InsertGitCommon(metrics model.GitCommonAttribute, statement dbstatement.DBStatement)
 	Close()
 }
 
@@ -63,7 +66,7 @@ func NewDBClient(conf *config.Config) (DBInterface, error) {
 		}
 		return nil, err
 	}
-	tables := []DBStatement{kubvizTable, rakeesTable, kubePugDepricatedTable, kubepugDeletedTable, ketallTable, outdateTable, clickhouseExperimental, containerDockerhubTable, containerGithubTable, gitTable, kubescoreTable, dockerHubBuildTable}
+	tables := []DBStatement{kubvizTable, rakeesTable, kubePugDepricatedTable, kubepugDeletedTable, ketallTable, outdateTable, clickhouseExperimental, containerDockerhubTable, containerGithubTable, kubescoreTable, dockerHubBuildTable, DBStatement(dbstatement.AzureDevopsTable), DBStatement(dbstatement.GithubTable), DBStatement(dbstatement.GitlabTable), DBStatement(dbstatement.BitbucketTable), DBStatement(dbstatement.GiteaTable)}
 	for _, table := range tables {
 		if err = splconn.Exec(context.Background(), string(table)); err != nil {
 			return nil, err
@@ -415,3 +418,27 @@ func (c *DBClient) InsertContainerEventGithub(event string) {
 		log.Fatal(err)
 	}
 }
+
+func (c *DBClient) InsertGitCommon(metrics model.GitCommonAttribute, statement dbstatement.DBStatement) {
+	var (
+		tx, _   = c.conn.Begin()
+		stmt, _ = tx.Prepare(string(statement))
+	)
+	defer stmt.Close()
+	if _, err := stmt.Exec(
+		metrics.RepositoryName,
+		metrics.Author,
+		metrics.GitProvider,
+		metrics.CommitID,
+		metrics.CommitUrl,
+		metrics.EventType,
+		metrics.RepoName,
+		metrics.TimeStamp,
+		metrics.Event,
+	); err != nil {
+		log.Fatal(err)
+	}
+	if err := tx.Commit(); err != nil {
+		log.Fatal(err)
+	}
+}

client/pkg/clickhouse/statements.go

@@ -90,6 +90,7 @@ const dockerHubBuildTable DBStatement = `
 		Event String
 	) engine=File(TabSeparated)
 	`
+
 const InsertDockerHubBuild DBStatement = "INSERT INTO dockerhubbuild (PushedBy, ImageTag, RepositoryName, DateCreated, Owner, Event) VALUES (?, ?, ?, ?, ?, ?)"
 const InsertRakees DBStatement = "INSERT INTO rakkess (ClusterName, Name, Create, Delete, List, Update) VALUES (?, ?, ?, ?, ?, ?)"
 const InsertKetall DBStatement = "INSERT INTO getall_resources (ClusterName, Namespace, Kind, Resource, Age) VALUES (?, ?, ?, ?, ?)"
@@ -99,6 +100,5 @@ const InsertDeletedApi DBStatement = "INSERT INTO DeletedAPIs (ClusterName, Obje
 const InsertKubvizEvent DBStatement = "INSERT INTO events (ClusterName, Id, EventTime, OpType, Name, Namespace, Kind, Message, Reason, Host, Event, FirstTime, LastTime) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
 const clickhouseExperimental DBStatement = `SET allow_experimental_object_type=1;`
 const containerDockerhubTable DBStatement = `CREATE table IF NOT EXISTS container_dockerhub(event JSON) ENGINE = MergeTree ORDER BY tuple();`
-const gitTable DBStatement = `CREATE table IF NOT EXISTS git_json(event JSON) ENGINE = MergeTree ORDER BY tuple();`
 const containerGithubTable DBStatement = `CREATE table IF NOT EXISTS container_github(event JSON) ENGINE = MergeTree ORDER BY tuple();`
 const InsertKubeScore string = "INSERT INTO kubescore (id, namespace, cluster_name, recommendations) VALUES (?, ?, ?, ?)"