Простейшее использование Elasticsearch для работы с журналом регистрации

input {
  http {
    tags => "onec"
    port => 5050
    host => "0.0.0.0"
 }
}

filter {
 if "onec" in [tags]{
  mutate {
    convert => {
      "Port"                              => "integer"
      "SyncPort"                          => "integer"
      "Session"                           => "integer"
      "Connection"                        => "integer"      

      "Level"                             => "string"
      "User"                              => "string"
      "Computer"                          => "string"
      "ApplicationName"                   => "string"
      "Event"                             => "string"
      "Comment"                           => "string"
      "Metadata"                          => "string"
      "Data"                              => "string"
      "DataPresentation"                  => "string"
      "UserName"                          => "string"
      "ApplicationPresentation"           => "string"
      "EventPresentation"                 => "string"
      "MetadataPresentation"              => "string"
      "TransactionStatus"                 => "string"
      "ServerName"                        => "string"  
      "TransactionID"                     => "string"    
    }
  }

   #События транзакци многочисленны и не информативны, удалим их
  if "_$Transaction$_" in [Event] {
     drop{ }
  }

  date {
    match => [ "Date", "yyyy-MM-dd HH:mm:ss", "ISO8601" ]
    timezone => "Europe/Moscow"
    target => "Date"
  }
}
}

output {
  if "onec" in [tags] {
   elasticsearch {
    hosts => [ "127.0.0.1:9200" ]
    index => "oncjornallog-%{+YYYY.MM}"
   }
  }
  #очень удобный вывод для отладочных работ
  #  stdout {}
  #file {
  #  path => "c:\test\logstash.log"
  #}
}

#Область РаботаСПериодомВыгрузкиЖР

Функция КлючХранилищаДатыНижнейГраницыВыгрузкиЖР()
	Возврат "elasticLogstashLastDateOfLogData";
КонецФункции


Функция ПолучитьДатуНижнейГраницыВыгрузкиЖР() Экспорт
	ключ = КлючХранилищаДатыНижнейГраницыВыгрузкиЖР();
	датаГраницыВыгрузки = ХранилищеОбщихНастроек.Загрузить(
			ключ, 
			ключ,
			Неопределено,
			""
	);
		
	Если НЕ ЗначениеЗаполнено(датаГраницыВыгрузки) Тогда
		//т.к. конечная дата выгрузки ЖР смещается на час относительно
		//нижней границы выгрузки, то первую нижнюю границу устанавливаем
		//вниз на три месяца от текущей даты
		датаГраницыВыгрузки = ДобавитьМесяц(ТекущаяДата(),-3);
	КонецЕсли;
	
	Возврат датаГраницыВыгрузки;
КонецФункции	


Процедура СохранитьДатуНижнейГраницыВыгрузкиЖР(датаГраницыВыгрузки) Экспорт
	ключ = КлючХранилищаДатыНижнейГраницыВыгрузкиЖР();
	ХранилищеОбщихНастроек.Сохранить(
			ключ,
			ключ,
			датаГраницыВыгрузки,
			Неопределено,
			""
	);
КонецПроцедуры
	
	
Функция ИнициироватьПериодВыгрузки(знач период = 3600)
	//инициируем период выгрузки
	датаНачала = ПолучитьДатуНижнейГраницыВыгрузкиЖР();
	датаОкончания = ТекущаяДата();
	//если выгружаемый период больше часа
	//смещаем дату максимальной границы выгрузки на час
	//это ограничит максимальный объем выгружаемых за один раз данных
	Если период <=0 Тогда
		период = 3600;
	КонецЕсли;
	датаОкончания = ?(датаОкончания<(датаНачала+3600), датаОкончания, датаНачала+период);
	
	Возврат Новый Структура("ДатаНачала, ДатаОкончания", датаНачала, датаОкончания);
КонецФункции
#КонецОбласти

Функция ОтправитьЖРвLogstash() Экспорт
	настройки = ПолучитьНастройкиДоступаКElastic();
	
	//Тест соединения 	
	ответ = ОтправитьPOSTЗапрос(настройки, "");
	Если ответ.КодСостояния <> 200 Тогда
		ЗаписьЖурналаРегистрации("_Elastic_", 
					УровеньЖурналаРегистрации.Ошибка, ,
					"Логи ЖР. Не удалось установить соединение.", 
					СтрокаСоединенияИнформационнойБазы()
				);
		Возврат "Не удалось установить соединение.";
	КонецЕсли;	
	
	Попытка
		//ограничиваем период выгружаемых данных текущим моментом
		//это не строгое ограничени, т.к. реально выгрузка будет остановлена
		//когда начало выбираемого периода будет больше указанной даты,
		//а значит верхняя граница будет заведомо выше данного ограничения
		//это допустимо, т.к. именно реальная верхняя граница периода 
		//будет сохранена как начало периода отбора для следующего запуска 
		//выгрузки		
		датаВерхнейГраницыВыгрузки = ТекущаяДата();
		соотИменКолонок = ПолучитьСоответствиеИменКолонокЖР();
		период = ИнициироватьПериодВыгрузки(настройки.Период);
		
		Пока период.ДатаНачала <= датаВерхнейГраницыВыгрузки Цикл
			табЗаписейЖР = ВыбратьДанныеЖРЗаПериод(период.ДатаНачала, период.ДатаОкончания);
			Если табЗаписейЖР.Количество() = 0 Тогда
				СохранитьДатуНижнейГраницыВыгрузкиЖР(период.ДатаОкончания);
			Иначе
				ОтправитьДанныеЗаПериод(настройки, табЗаписейЖР);	
			КонецЕсли;
			период = ИнициироватьПериодВыгрузки(настройки.Период); 
		КонецЦикла;
	Исключение
		ошибка = ОписаниеОшибки();
		ЗаписьЖурналаРегистрации("_Elastic_", 
					УровеньЖурналаРегистрации.Ошибка, ,
					"Логи ЖР. Не удалось выгрузить данные ЖР.", 
					СтрокаСоединенияИнформационнойБазы()+" "+ошибка
		);
		ВызватьИсключение ошибка;
	КонецПопытки;
КонецФункци