Migrate from correlation service to direct Logstash connection in office365 integration.

Author: JocLRojas

office365/configuration/const.go

@@ -14,7 +14,8 @@ const (
 	endPointStartSubscription = "/activity/feed/subscriptions/start"
 	endPointContent           = "/activity/feed/subscriptions/content"
 	BASEURL                   = "https://manage.office.com/api/v1.0/"
-	CORRELATIONURL            = "http://correlation:8080/v1/newlog"
+	LogstashEndpoint          = "http://%s:%s"
+	UTMLogSeparator           = "<utm-log-separator>"
 )
 
 func GetInternalKey() string {
@@ -36,3 +37,11 @@ func GetStartSubscriptionLink(tenant string) string {
 func GetContentLink(tenant string) string {
 	return fmt.Sprintf("%s%s%s", BASEURL, tenant, endPointContent)
 }
+
+func GetLogstashHost() string {
+	return utils.Getenv("UTM_LOGSTASH_HOST")
+}
+
+func GetLogstashPort() string {
+	return utils.Getenv("UTM_LOGSTASH_PORT")
+}

office365/processor/processor.go

@@ -146,9 +146,9 @@ func (o *OfficeProcessor) GetLogs(startTime string, endTime string, group types.
 				if len(details) > 0 {
 					logsCounter += len(details)
 					cleanLogs := ETLProcess(details, group)
-					err = SendToCorrelation(cleanLogs)
+					err := SendToLogstash(cleanLogs)
 					if err != nil {
-						utils.Logger.ErrorF("error sending logs to correlation: %v", err) // Debug
+						utils.Logger.ErrorF("error sending logs to logstash: %v", err) // Debug
 						continue
 					}
 				}

office365/processor/sendData.go

@@ -1,35 +1,71 @@
 package processor
 
 import (
+	"bytes"
+	"crypto/tls"
 	"encoding/json"
+	"fmt"
 	"net/http"
+	"strings"
+	"time"
 
+	"github.com/threatwinds/logger"
 	"github.com/utmstack/UTMStack/office365/configuration"
 	"github.com/utmstack/UTMStack/office365/utils"
 )
 
-func SendToCorrelation(data []TransformedLog) error {
-	utils.Logger.Info("uploading %d logs...", len(data))
+var transport = &http.Transport{
+	MaxIdleConns:          100,
+	IdleConnTimeout:       2 * time.Second,
+	ResponseHeaderTimeout: 2 * time.Second,
+	ForceAttemptHTTP2:     true,
+	TLSClientConfig: &tls.Config{
+		InsecureSkipVerify: true,
+	},
+}
+
+var client = &http.Client{Transport: transport, Timeout: 2 * time.Second}
 
+func SendToLogstash(data []TransformedLog) *logger.Error {
+	var logStrings []string
 	for _, log := range data {
 		body, err := json.Marshal(log)
 		if err != nil {
 			utils.Logger.ErrorF("error encoding log to JSON: %v", err)
 			continue
 		}
+		logStrings = append(logStrings, string(body))
+	}
 
-		_, status, e := utils.DoReq[map[string]interface{}](configuration.CORRELATIONURL, body, http.MethodPost, map[string]string{})
-		if e != nil {
-			utils.Logger.ErrorF("error sending log to correlation engine: %v", e)
-			continue
-		} else if status != http.StatusOK && status != http.StatusCreated {
-			utils.Logger.ErrorF("error sending log to correlation engine: status code %d", status)
-			continue
+	if len(logStrings) == 0 {
+		return nil
+	}
+
+	var logs string
+	for _, str := range logStrings {
+		logs += str + configuration.UTMLogSeparator
+	}
+
+	url := fmt.Sprintf(configuration.LogstashEndpoint, configuration.GetLogstashHost(), configuration.GetLogstashPort())
+
+	req, err := http.NewRequest("POST", url, bytes.NewBufferString(logs))
+	if err != nil {
+		return utils.Logger.ErrorF("error creating request: %v", err.Error())
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		if !strings.Contains(err.Error(), "Client.Timeout exceeded while awaiting headers") {
+			utils.Logger.ErrorF("error sending logs with error: %v", err.Error())
 		}
+		return utils.Logger.ErrorF("error sending logs: %v", err.Error())
+	}
+	defer resp.Body.Close()
 
-		utils.Logger.Info("log successfully sent to correlation engine")
+	if resp.StatusCode != http.StatusOK {
+		return utils.Logger.ErrorF("error sending logs with http code %d", resp.StatusCode)
 	}
 
-	utils.Logger.Info("all logs were sent to correlation")
+	utils.Logger.Info("successfully sent %d logs to Logstash", len(logStrings))
 	return nil
 }