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

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

Работа была сделана для устранения проблем долгого поиска информации "кто виноват" и т.п., в базах РИБ по журналам регистрации. Все наработки находятся в git: https://github.com/smilut/uploadRegJornal2Elastic.git

Разработка и тестирование велись на платформе 8.3.13.1513. Функционал запуска по расписанию тестировался в конфигурации Бухгалтерия предприятия 3.0.70.50.

Первоначально хотелось использовать filebeat, но после некоторых попыток прямого обращения к файлам журнала регистрации, показалось более простым реализовать "filebeat" в виде обработки с запуском по регламентному заданию. Данные от обработки принимаются в logstah и после небольшой трансформации помещаются в индекс Elasticsearch.

Описание файлов git:

ОбработкаВыгрузкиЖРсЗапускомПоРасписанию.epf — обработка, которая загружается как дополнительная обработка и запускается по расписанию (функционал БСП). В этой обработке указываются параметры подключения к logstash для выгрузки данных и параметры подключения к Elastic для выполнения запросов. Обработка выполнения запросов еще в вяло текущей разработке, т.к. ее наличие не критично для поиска данных. Скорее обработка формирования запросов необходима пользователям, которые не хотят менять привычный интерфейс 1С на что-то другое. Добавляет к данным журнала регистрации реквизит с указанием строки подключения к базе, из которой выгружается журнал регистрации.

export.json — настроенные в нашей системе дашборды и сопутствующие объекты.

oncjornallog_template.json — образец запроса создания шаблона для индекса журнала. В шаблон добавлен анализатор, который разбивает содержимое некоторых полей на лексемы, что позволяет искать данные по не полностью набранному слову. С настройками по-умолчанию такого поведения добиться не удалось.

oneclog.conf — файл конфигурации logstash, при условии что сервис принимает данные только от 1С, если надо завести данные от нескольких систем настройка немного усложняется. В своих настройках мы разделяли обработку данных используя ключ "tag".

oneclog.conf для приема данных с нескольких источников

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: estlogstash.log"
#}
}

Обработка выгрузки журнала построена на том же принципе, что и filebeat, с небольшим ограничением или усовершенствованием. При запуске запуске можно указать нижнюю границу загрузки, которая будет смещаться при каждой успешной передаче данных в Elastik. Если дата начала выгрузки не указана, то отступаем на три месяца от текущей даты.

Код работы с периодом выгрузки журнала регистрации

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

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


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

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

Возврат датаГраницыВыгрузки;
КонецФункции


Процедура СохранитьДатуНижнейГраницыВыгрузкиЖР(датаГраницыВыгрузки) Экспорт
ключ = КлючХранилищаДатыНижнейГраницыВыгрузкиЖР();
ХранилищеОбщихНастроек.Сохранить(
ключ,
ключ,
датаГраницыВыгрузки,
Неопределено,
""
);
КонецПроцедуры


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

Возврат Новый Структура("ДатаНачала, ДатаОкончания", датаНачала, датаОкончания);
КонецФункции
#КонецОбласти

Конечно же существует возможность задвоения переданных данных, например процесс передал порцию данных и упал до обработки ответа (выключение питания, перезагрузка и т.д.). В этом случае граница чтения данных не сдвинется и данные будут переданы повторно, но в данном случае это не критично. Elastic позиционируется, как система быстрого поиска и если в результате поиска будут найдены две абсолютно идентичные записи в логах это не исказит описанного в них события. Filebeat, насколько смог понять документацию, ведет себя также. В своих "недрах" он сохраняет указатель до какого момента вычитан файл лога, что бы в случае сбоя начать с этой позиции. Просто за счет скорости опроса у filebeat (по-умолчанию, кажется 30ms) порция передаваемых данных, как правило очень маленькая и задвоение маловероятно. Необходимая периодичность выгрузки в 1С зависит от интенсивности работы пользователей, обменов, регламентных заданий и т.п. В тестовых системах устанавливали по 2 минуты.

Код выгрузки данных журнала регистрации в Elastic

Функция ОтправитьЖРвLogstash() Экспорт
настройки = ПолучитьНастройкиДоступаКElastic();

//Тест соединения
ответ = ОтправитьPOSTЗапрос(настройки, "");
Если ответ.КодСостояния <> 200 Тогда
ЗаписьЖурналаРегистрации("_Elastic_",
УровеньЖурналаРегистрации.Ошибка, ,
"Логи ЖР. Не удалось установить соединение.",
СтрокаСоединенияИнформационнойБазы()
);
Возврат "Не удалось установить соединение.";
КонецЕсли;

Попытка
//ограничиваем период выгружаемых данных текущим моментом
//это не строгое ограничени, т.к. реально выгрузка будет остановлена
//когда начало выбираемого периода будет больше указанной даты,
//а значит верхняя граница будет заведомо выше данного ограничения
//это допустимо, т.к. именно реальная верхняя граница периода
//будет сохранена как начало периода отбора для следующего запуска
//выгрузки
датаВерхнейГраницыВыгрузки = ТекущаяДата();
соотИменКолонок = ПолучитьСоответствиеИменКолонокЖР();
период = ИнициироватьПериодВыгрузки(настройки.Период);

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

Если кому-то лень качать с git прикладываю архив файлов в текущем состоянии разработки.

Буду рад замечаниям, критике указаниям на ошибки и вопросам, даже если не смогу на них ответить ).

7 Comments

goleaff2006 21.06.2019 at 17:50

она выгружает за период? Как определяет что уже выгружала а что нет?

Reply ↓
milut 21.06.2019 at 21:57

В настройках сохраняется граница выгруженных данных. С этого момента начинается следующая выгрузка.

Reply ↓
milut 22.06.2019 at 17:06

(1) Спасибо. Дописал ответ в статью.

Reply ↓
malikov_pro 08.09.2019 at 10:02

Зачем logstash, если подготовку данных можно сделать на уровне выгрузки?

Reply ↓
milut 09.09.2019 at 09:11

(4) Использовали elastic community, в котором есть ограничения настройки прав доступа. Поэтому прямой доступ не открывали. Никаких особенных причин для использования logstash не было.

Reply ↓

malikov_pro 09.09.2019 at 14:17

Дату последней записи в El можно получить

{
«size»: 1,
«sort»: { «Дата»: «desc»},
«query»: {
«term»: {
«ConnectionString»: «FILE=»E:\1C\BASE\DEMOTRD»;»
}
}
}

Показать

Для корректного поиска нужен маппинг

«ConnectionString»: {
«type»: «keyword»,
«fields»: {
«keyword»: {
«type»: «keyword»,
«ignore_above»: 256
}
}
},

Показать

И проверку с ЖР сделать

Если СоответствиеДанныеИндекса.Количество() = 0 Тогда
ЭтоНачалоДанных = Истина;
ИначеЕсли СоответствиеДанныеИндекса.Получить(«Дата») <> Стр_ЖР.Дата Тогда
ЭтоНачалоДанных = Истина;
ИначеЕсли
СоответствиеДанныеИндекса.Получить(«Событие») = Стр_ЖР.Событие
И СоответствиеДанныеИндекса.Получить(«Сеанс») = Стр_ЖР.Сеанс
И СоответствиеДанныеИндекса.Получить(«ПредставлениеДанных») = Стр_ЖР.ПредставлениеДанных
Тогда

ЭтоНачалоДанных = Истина;
ТочноеСовпадение = Истина;
КонецЕсли;

Показать

ELK узлы спокойно прикрываются BasicAuth от NGINX + HTTPS

Reply ↓

milut 09.09.2019 at 22:54

(6) наверняка можно и так. Мы смотрели в сторону nginx, но как ограничить доступ только передачей данных не нашли. Всегда оставался доступ к api. А крайнюю дату определял на стороне 1С только по субъективным причинам, тем способом который показался наиболее подходящим.

Reply ↓

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

Related Posts

7 Comments

Leave a Comment Отменить ответ