Реализована пакетная отправка данных в эластики соответствующие настройки. Произведена чистка кода

log-processor.go

@@ -1,27 +1,12 @@
 // ---------------------------------------------------------------------- 
 // Программа для анализа и записи журналов 1С
-// Версия: 1.0.0
+// Версия: 1.2.0
-// Автор: Сергей Калинин
+// Автор: Сергей Калинин, svk@nuk-svk.ru
+// https://git.nuk-svk.ru/svk/1c-logprocessor
+// Лицензия: GPLv3
 // ---------------------------------------------------------------------- 
 // Использование log-processor:
-// -dict-file string
+// см. README.md
-//   Файл со словарём (DICT_FILE) (default "1Cv8.lgf")
-// -dir-in string
-//   Каталог для исходных файлов (DIR_IN) (default "in")
-// -dir-log string
-//   Каталог для лога работы (DIR_LOG) (default "log")
-// -dir-out string
-//   Каталог для обработанных файлов (DIR_OUT) (default "out")
-// -dir-temp string
-//   Каталог для временных файлов (TEMP) (default "/tmp")
-// -log-file-ext string
-//   Расширение файлов с журналами (обрабатываемых логов) (LOG_FILE_EXT) (default ".lgp")
-// -object-types string
-//   Список типов объектов для выборки разделённый запятой (OBJECT_TYPES) (default "1,2,3,4,5,6,7,8")
-// -out-format string
-//   Выводить данные в формате JSON (OUT_FORMAT) (default "json")
-// -send-to-es
-//   Отправлять данные в ElasticSearch (SEND_TO_ES)
 // -------------------------------------------------------------------------
 // ELASTICSEARCH_URL="https://user:pass@elastic:9200"
 
@@ -54,6 +39,11 @@ import (
     opensearchapi "github.com/opensearch-project/opensearch-go/v2/opensearchapi"
    "github.com/kardianos/service"
 )
+
+const (
+    Version = "1.2.0"
+)
+
 var (
     DirCfg         string
     FileCfg        string
@@ -91,6 +81,7 @@ var (
     InFile         string
     WorkLogOut     string
     ServiceName    string
+    // PrintVersion   bool
 )
 
 type Config struct {
@@ -108,8 +99,8 @@ type Config struct {
     EsPassword       string `ini:"esPassword"`
     EsIndexPrefix    string `ini:"esIndexPrefix"`
     EsSkipVerify     bool   `ini:"esSkipVerify"`
-    // EsBulk           bool   `ini:"esBulk"`
+    EsBulk           bool   `ini:"esBulk"`
-    // EsBulkRecordsQuantity int  `ini:"esBulkQuantity"`
+    EsBulkRecordsQuantity int  `ini:"esBulkQuantity"`
     OutFormat        string `ini:"outFormat"`
     WriteOutFile     bool   `ini:"writeOutFile"`
     TimeZoneOffset   string `ini:"timeZoneOffset"`
@@ -251,6 +242,10 @@ esPassword=user_password
 esSkipVerify=false
 ; Перфикс индекса в ElasticSearch
 esIndexPrefix=test_log
+; Пакетная вставка данных
+esBulk=true
+; Количество записей в одном "пакете"
+esBulkRecordsQuantity=10
 
 [Processing]
 ; Включение вывода обработанной информации в файл
@@ -871,6 +866,11 @@ func runOperations () {
     var client *opensearch.Client
 
     if InFile != "" {
+        if SendToEs {
+            client = esConnect()
+            EsIndexName = EsIndexPrefix + "-" + getDate(InFile)
+            esCreateIndex(EsIndexName, client)
+        }
         tail(client, InFile, os.Stdout)
         return
     }
@@ -936,9 +936,10 @@ func tail(client *opensearch.Client, fileName string, out io.Writer) {
         recBegin = false
         outLine string
         l string
-        // bulkCount int
+        bulkCount int
-        // arrRecords []map[string]string
+        arrRecords []map[string]string
         fOut *os.File
+        indexName string
         // matchedEnd bool
     )
 
@@ -992,11 +993,6 @@ func tail(client *opensearch.Client, fileName string, out io.Writer) {
     // Регулярные выражения для определения начала и конца записи
     regexpBeginRecord := regexp.MustCompile(`^\{\d{14},\w{1},$`)
     regexpEndRecord := regexp.MustCompile(`^\},$`)
-      // var (й
-          // isPrefix bool = true
-          // // err error = nil
-          // line []byte
-      // )
     for {
         line, err := Readln(r)
         // for line, _, err := r.ReadLine(); err != io.EOF; line, _, err = r.ReadLine() {
@@ -1031,6 +1027,7 @@ func tail(client *opensearch.Client, fileName string, out io.Writer) {
                 result := parseString(outLine, logOffset)
                 if Debug {
                     log.Println(len(result), result)
+                    log.Println("Префикс индекса:", EsIndexPrefix, result["index_day_prefix"])
                 }
                 // Пишем выходные файлы
                 if WriteOutFile {
@@ -1042,25 +1039,30 @@ func tail(client *opensearch.Client, fileName string, out io.Writer) {
                 }
                 // Отправляем данные в ElasticSearch / OpenSearch
                 if SendToEs {
-                    // if result["index_day_prefix"] == "" {
+                    if result["index_day_prefix"] != "" {
-                        // continue
+                        indexName = EsIndexPrefix + "-" + result["index_day_prefix"]
-                    // }
-                    indexName := EsIndexPrefix + "-" + result["index_day_prefix"]
-                    // пакетная отправка данных в эластик
-                    // addRes := esAddRecord(client, indexName, result)
-                    if Debug {
-                        log.Println("Index name:", indexName)
                     }
-                    go esAddRecord(client, indexName, result)
+                    // if Debug {
-                    time.Sleep(TailSleep)
+                        // log.Println("Index name:", indexName)
-                    // time.Sleep(5 * time.Millisecond)
+                        // // log.Println("Пакетная обработка:", EsBulk)
-
-                    // Если запись не добавилась делаем повторную попытку через 5 сек.
-                    // if !addRes {
-                        // time.Sleep(5 * time.Second)
-                        // addRes = esAddRecord(client, indexName, result)
-                        // log.Println("Повторная попытка добавления записи:", result, addRes)
                     // }
+                    // Пакетная обработка, запускается если выставлена опция командной строки
+                    // иначе отправляется по одной строке
+                    if EsBulk {
+                        if bulkCount < EsBulkRecordsQuantity {
+                            if len(result) != 0 {
+                                arrRecords = append(arrRecords, result)
+                                bulkCount++
+                            }
+                        } else {
+                            go esAddRecordsBulk(client, indexName, arrRecords)
+                            bulkCount = 0
+                            arrRecords = nil
+                        }
+                    } else {
+                        go esAddRecord(client, indexName, result)
+                    }
+                    time.Sleep(TailSleep)
                 }
                 writeTempFile(tempFile, logOffset)
                 if Debug {
@@ -1117,7 +1119,14 @@ func tail(client *opensearch.Client, fileName string, out io.Writer) {
                 }
             }
         }
+        // Если получен сигнал о наличии новго файла то прекращаем читать старый
+        // и отправляем пакет данных в эластик (чтобы не потерять последние записи)
         if shutUp {
+            if SendToEs && EsBulk {
+                go esAddRecordsBulk(client, indexName, arrRecords)
+                bulkCount = 0
+                arrRecords = nil
+            }
             break
         }
     }
@@ -1198,6 +1207,9 @@ func esCreateIndex(indexName string, client *opensearch.Client) {
         log.Println("failed to create index ", err, createIndexResponse)
         os.Exit(1)
     }
+    if Debug {
+        log.Println("Create index:", createIndex)
+    }
     fmt.Println(createIndexResponse.StatusCode)
     if strings.Contains(createIndexResponse.String(), "already exists") {
         log.Println(createIndexResponse.IsError())
@@ -1207,35 +1219,6 @@ func esCreateIndex(indexName string, client *opensearch.Client) {
     }
 }
 
-func esAddRecordsBulk(client *opensearch.Client, indexName string, arrLine []map[string]string) {
-    // Perform bulk operations.
-    for _, obj := range arrLine {
-        aJson, err := json.Marshal(obj)
-        fmt.Println(aJson)
-        if err != nil{
-		    log.Printf("Error encoding query: %s", err)
-        }
-        
-    }
-    // blk, err := client.Bulk(
-		// strings.NewReader(`
-    // { "index" : { "_index" : "go-test-index1", "_id" : "2" } }
-    // { "title" : "Interstellar", "director" : "Christopher Nolan", "year" : "2014"}
-    // { "create" : { "_index" : "go-test-index1", "_id" : "3" } }
-    // { "title" : "Star Trek Beyond", "director" : "Justin Lin", "year" : "2015"}
-    // { "update" : {"_id" : "3", "_index" : "go-test-index1" } }
-    // { "doc" : {"year" : "2016"} }
-// `),
-	// )
-// 
-    // if err != nil {
-        // fmt.Println("failed to perform bulk operations", err)
-        // os.Exit(1)
-    // }
-    // fmt.Println("Performing bulk operations")
-    // fmt.Println(blk)
-
-}
 
 // Вставка записи в индекс ES
 func esAddRecord(client *opensearch.Client, indexName string, line map[string]string) bool {
@@ -1273,13 +1256,108 @@ func esAddRecord(client *opensearch.Client, indexName string, line map[string]st
     } else {
         operResult = true
     }
-    if Debug {
+
-        log.Println(insertResponse)
+    // Проверям результат добавления записи, если код возврата не равен 200
+    // то повторяем несколько раз
+    if insertResponse.StatusCode == 200 && operResult {
+        log.Println("Пакет данных добавлен")
+    } else {
+        // Ждем две секунды перед повторной отправкой
+        j := 1
+        for j <= 5 {
+            time.Sleep(2 * time.Second)
+            insertResponse, err := req.Do(context.Background(), client)
+            if err != nil {
+                log.Println("Ошибка повторного добавления записи в opensearch (процедура esAddRecordsBulk):", err)
             }
+            if insertResponse.StatusCode == 200 {
+                operResult = true
+                break
+            }
+            j++
+        }
+    }
+
+    if Debug {
+        log.Printf("Ответ от %s при добавлении записи: %s", EsUrl, insertResponse)
+    }
+
     defer insertResponse.Body.Close()
     return operResult
 }
 
+// Пакетная Вставка записуй в индекс ES
+func esAddRecordsBulk(client *opensearch.Client, indexName string, arr []map[string]string) bool {
+    if Debug {
+        log.Println("================= Отправка пакета данных ===================")
+        log.Println("Индекс:", indexName)
+        // log.Println(arr)
+    }
+    if len(arr) == 0 {
+        return false
+    }
+    var (
+        operResult bool
+        resultJson string
+    )
+    resultJson = ""
+    for _, line := range arr {
+        // if Debug {
+            // log.Println("----", i, line)
+        // }
+        if len(line) != 0 {
+            resultJson += fmt.Sprintf("{\"create\": { \"_index\": \"%s\"}}\n", indexName)
+            aJson, err := json.Marshal(line)
+            if err != nil {
+                log.Printf("Error encoding: %s", line, "with error %s", err)
+            } else {
+                 resultJson += fmt.Sprintf("%s\n", string(aJson))
+            }
+        }
+    }
+    // _ = strings.NewReader(string(aJson))
+    if Debug {
+        log.Println(string(resultJson))
+    }
+
+    r := strings.NewReader(resultJson)
+    bulkResp, err := client.Bulk(r)
+    if err != nil {
+        log.Println("Ошибка добавления пакета данных в opensearch (процедура esAddRecordsBulk):", err)
+        operResult = false
+    } else {
+        operResult = true
+    }
+
+    // Проверям результат добавления записи, если код возврата не равен 200
+    // то повторяем несколько раз
+    if bulkResp.StatusCode == 200 && operResult {
+        log.Println("Пакет данных добавлен")
+    } else {
+        // Ждем две секунды перед повторной отправкой
+        j := 1
+        for j <= 5 {
+            time.Sleep(2 * time.Second)
+            bulkResp, err = client.Bulk(r)
+            if err != nil {
+                log.Println("Ошибка повторного добавления пакета данных в opensearch (процедура esAddRecordsBulk):", err)
+            }
+            if bulkResp.StatusCode == 200 {
+                operResult = true
+                break
+            }
+            j++
+        }
+    }
+    if Debug {
+        log.Printf("Ответ от %s при добавлении пакета данных: %s", EsUrl, bulkResp)
+    }
+
+    defer bulkResp.Body.Close()
+    return operResult
+}
+
+
 // Читаем  конфиг и определяем переменные
 func readConfigFile(fileName string) {
     cfg, err := ini.LoadSources(ini.LoadOptions{
@@ -1308,6 +1386,8 @@ func readConfigFile(fileName string) {
         EsPassword = cfg.Section("ElasticSearch").Key("esPassword").String()
         EsSkipVerify = cfg.Section("ElasticSearch").Key("esSkipVerify").MustBool()
         EsIndexPrefix = cfg.Section("ElasticSearch").Key("esIndexPrefix").String()
+        EsBulk = cfg.Section("ElasticSearch").Key("esBulk").MustBool()
+        EsBulkRecordsQuantity, _ = strconv.Atoi(cfg.Section("ElasticSearch").Key("esBulkRecordsQuantity").String())
     }
 
     WriteOutFile = cfg.Section("Processing").Key("writeOutFile").MustBool()
@@ -1402,8 +1482,8 @@ func main() {
     flag.StringVar(&EsPassword, "es-password", "", "Пароль пользователя Elastic Search (ELASTICSEARCH_PASSWORD)")
     flag.StringVar(&EsIndexPrefix, "es-index-prefix", "", "Префикс имени индекса Elastic Search (ELASTICSEARCH_INDEX_PREFIX)")
     flag.BoolVar(&EsDiscoverNode, "es-discover-node", false, "Получать список узлов кластера ElasticSearch")
-    // flag.IntVar(&EsBulkRecordsQuantity, "es-bulk-quantity", 10, "Количество записей в одном запросе для пакетной вставки")
+    flag.IntVar(&EsBulkRecordsQuantity, "es-bulk-quantity", 10, "Количество записей в одном запросе для пакетной вставки")
-    // flag.BoolVar(&EsBulk, "es-bulk", false, "Пакетная вставка записей в Elastic Search")
+    flag.BoolVar(&EsBulk, "es-bulk", false, "Пакетная вставка записей в Elastic Search")
     flag.BoolVar(&EsSkipVerify, "es-skip-verify", false, "Пропустить проверку сертификатов при подключении к Elastic Search")
     flag.BoolVar(&WriteOutFile, "write-out-file", false, "Запись обработанных данных в файл")
     flag.StringVar(&OutFormat, "out-format", "csv", "Формат данных на выходе (csv, json) (OUT_FORMAT)")
@@ -1413,6 +1493,12 @@ func main() {
     flag.StringVar(&Duration, "tail-sleep", "5ms", "Задержка чтения входного файла. Разрешены обозначения 'ns', 'us' ('µs'), 'ms', 's', 'm','h')")
     flag.StringVar(&InFile, "file", "", "Имя файла для обработки. Если требуется обработать один файл")
     flag.StringVar(&WorkLogOut, "worklog-out", "file", "Направление вывода журнала работы программы (console, file)")
+    // flag.BoolVar(&Version, "version", false, "Версия программы")
+    flag.BoolFunc("version", "Версия программы", func(s string) error {
+        fmt.Println("Версия:", Version)
+        os.Exit(0)
+        return nil
+    })
 
     flag.Parse()
 
@@ -1455,10 +1541,6 @@ func main() {
         LogFileExtention = os.Getenv("LOG_FILE_EXT")
     }
 
-   	// if os.Getenv("TEMP") != "" {
-	    // DirTemp = os.Getenv("TEMP")
-	// }
-	
     if os.Getenv("DIR_LOG") != "" {
         DirLog = os.Getenv("DIR_LOG")
     }