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

Author: JocLRojas

aws/configuration/const.go

@@ -3,8 +3,9 @@ package configuration
 import "github.com/utmstack/UTMStack/aws/utils"
 
 const (
-	CORRELATIONURL       = "http://correlation:8080/v1/newlog"
 	URL_CHECK_CONNECTION = "https://sts.amazonaws.com"
+	LogstashEndpoint     = "http://%s:%s"
+	UTMLogSeparator      = "<utm-log-separator>"
 )
 
 func GetInternalKey() string {
@@ -14,3 +15,11 @@ func GetInternalKey() string {
 func GetPanelServiceName() string {
 	return utils.Getenv("PANEL_SERV_NAME")
 }
+
+func GetLogstashHost() string {
+	return utils.Getenv("UTM_LOGSTASH_HOST")
+}
+
+func GetLogstashPort() string {
+	return utils.Getenv("UTM_LOGSTASH_PORT")
+}

aws/processor/pull.go

@@ -18,7 +18,7 @@ func PullLogs(startTime time.Time, endTime time.Time, group types.ModuleGroup) *
 		return err
 	}
 
-	err = SendToCorrelation(logs)
+	err = SendToLogstash(logs)
 	if err != nil {
 		return err
 	}

aws/processor/sendData.go

@@ -1,32 +1,71 @@
 package processor
 
 import (
+	"bytes"
+	"crypto/tls"
 	"encoding/json"
+	"fmt"
 	"net/http"
+	"strings"
+	"time"
 
 	"github.com/threatwinds/logger"
 	"github.com/utmstack/UTMStack/aws/configuration"
 	"github.com/utmstack/UTMStack/aws/utils"
 )
 
-func SendToCorrelation(data []TransformedLog) *logger.Error {
+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 %v", 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()
 
+	if resp.StatusCode != http.StatusOK {
+		return utils.Logger.ErrorF("error sending logs with http code %d", resp.StatusCode)
 	}
 
+	utils.Logger.Info("successfully sent %d logs to Logstash", len(logStrings))
 	return nil
 }