Аналитика логов Nginx с помощью Amazon Athena и Cube.js

Обикновено за мониторинг и анализ на работата на Nginx се използват търговски продукти или готови open-source алтернативи, като Prometheus + Grafana. Това е добър вариант за мониторинг или реално време аналитика, но не е особено удобно за исторически анализ. На който и да е популярен ресурс обемът от данни в логовете на nginx бързо нараства и за анализа на големи обеми данни е логично да се използва нещо по-специализирано.

В тази статия ще опиша как може да се използва Athena за анализ на логовете, взимайки Nginx за пример, и ще покажа как от тези данни да се създаде аналитичен дашборд, използвайки open-source фреймуърка cube.js. Ето пълната архитектура на решението:

TL:DR;
Връзка към готовия дашборд.

За събиране на информацията използваме Fluentd, за обработка — AWS Kinesis Data Firehose и AWS Glue, за съхранение — AWS S3. С помощта на тази комбинация можем да съхраняваме не само логовете на nginx, но и други събития, както и логовете на други услуги. Можете да замените някои части с аналогични на вашия стек, например, можете да пишете логовете в kinesis директно от nginx, заобикаляйки fluentd, или да използвате logstash за това.

Събиране на логовете на Nginx

По подразбиране логовете на Nginx изглеждат така:

4/9/2019 12:58:17 PM1.1.1.1 - - [09/Apr/2019:09:58:17 +0000] "GET /sign-up HTTP/2.0" 200 9168 "https://example.com/sign-in" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36" "-"
4/9/2019 12:58:17 PM1.1.1.1 - - [09/Apr/2019:09:58:17 +0000] "GET /sign-in HTTP/2.0" 200 9168 "https://example.com/sign-up" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36" "-"

Могат да бъдат анализирани, но много по-лесно е да се промени конфигурацията на Nginx, за да издава логовете в JSON:

log_format json_combined escape=json '{ "created_at": "$msec", '
            '"remote_addr": "$remote_addr", '
            '"remote_user": "$remote_user", '
            '"request": "$request", '
            '"status": $status, '
            '"bytes_sent": $bytes_sent, '
            '"request_length": $request_length, '
            '"request_time": $request_time, '
            '"http_referrer": "$http_referer", '
            '"http_x_forwarded_for": "$http_x_forwarded_for", '
            '"http_user_agent": "$http_user_agent" }';

access_log  /var/log/nginx/access.log  json_combined;

S3 за съхранение

За да съхраняваме логовете, ще използваме S3. Това позволява съхраняването и анализа на логовете на едно място, тъй като Athena може да работи с данните в S3 директно. По-нататък в статията ще опиша как правилно да съхраняваме и обработваме логовете, но за начало ни е необходим чист бакет в S3, в който да не се съхраняват други неща. Струва си предварително да обмислите в кой регион ще създадете бакета, защото Athena не е достъпна във всички региони.

Създаваме схема в конзолата Athena

Ще създадем таблица в Athena за логовете. Тя е необходима както за запис, така и за четене, ако планирате да използвате Kinesis Firehose. Отворете конзолата Athena и създайте таблица:

SQL за създаване на таблица

CREATE EXTERNAL TABLE `kinesis_logs_nginx`(
  `created_at` double, 
  `remote_addr` string, 
  `remote_user` string, 
  `request` string, 
  `status` int, 
  `bytes_sent` int, 
  `request_length` int, 
  `request_time` double, 
  `http_referrer` string, 
  `http_x_forwarded_for` string, 
  `http_user_agent` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.orc.OrcSerde' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.orc.OrcInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat'
LOCATION
  's3://'
TBLPROPERTIES ('has_encrypted_data'='false');

Създаваме Kinesis Firehose Stream

Kinesis Firehose ще запише данните, получени от Nginx, в S3 в избрания формат, разделяйки ги по директории в формат ГГГГ/ММ/ДД/ЧЧ. Това ще бъде полезно при четене на данните. Можете, разбира се, да записвате директно в S3 от fluentd, но в този случай ще трябва да пишете JSON, което не е ефективно поради големия размер на файловете. Освен това, при използването на PrestoDB или Athena, JSON е най-бавният формат на данни. Така че отворете конзолата Kinesis Firehose, натиснете „Create delivery stream“, изберете „direct PUT“ в полето „delivery“:

На следващата страница изберете „Record format conversion“ — „Enabled“ и изберете „Apache ORC“ като формат за запис. Според изследвания на някои Оуен О’Маллей, това е оптималният формат за PrestoDB и Athena. Като схема посочете таблицата, която създадохме по-горе. Обърнете внимание, че местоположението S3 в kinesis може да бъде всяко, от таблицата се използва само схемата. Но ако посочите друго местоположение S3, няма да можете да четете тези записи от таблицата.

Избираме S3 за съхранение и бакета, който създадохме по-рано. Aws Glue Crawler, за който ще разкажа по-късно, не може да работи с префикси в S3 бакета, така че е важно да го оставите празен.

Останалите опции могат да се променят в зависимост от вашите натоварвания, аз обикновено използвам стандартните. Обърнете внимание, че компресията в S3 не е налична, но ORC използва собствена компресия по подразбиране.

Fluentd

Сега, когато сме настроили съхранението и получаването на логове, трябва да настроим изпращането. Ще използваме Fluentd, защото обичам Ruby, но можете да използвате Logstash или да изпращате логовете в kinesis директно. Серверът Fluentd може да бъде пуснат по няколко начина, ще разкажа за docker, защото е просто и удобно.

Първо, ни трябва файлът за конфигурация fluent.conf. Създайте го и добавете source:

тип forward
порт 24224
свързване 0.0.0.0

Сега можете да стартирате сървър Fluentd. Ако имате нужда от по-сложна конфигурация, на Docker Hub има подробен гайд, включително как да създадете свой образ.

$ docker run 
  -d 
  -p 24224:24224 
  -p 24224:24224/udp 
  -v /data:/fluentd/log 
  -v :/fluentd/etc fluentd 
  -c /fluentd/etc/fluent.conf
  fluent/fluentd:stable

Тази конфигурация използва пътя /fluentd/log за кеширане на логовете преди изпращане. Може да минете и без това, но при рестартиране можете да загубите всички кеширани данни, за които сте работили усърдно. Портът също може да бъде всякакъв, 24224 е подразбиращият се порт на Fluentd.

Сега, когато имаме стартиран Fluentd, можем да изпратим логовете на Nginx там. Обикновено стартираме Nginx в Docker контейнер, и в този случай Docker има вградени драйвери на логовете за Fluentd:

$ docker run 
--log-driver=fluentd 
--log-opt fluentd-address=
--log-opt tag="{{.Name}}" 
-v /some/content:/usr/share/nginx/html:ro 
-d 
nginx

Ако стартирате Nginx по друг начин, можете да използвате лог файловете, в Fluentd има file tail plugin.

Нека добавим в конфигурацията Fluent парсинг на логовете, настройките отгоре:

@type parser
  key_name log
  emit_invalid_record_to_error false
  
    @type json

И изпращане на логовете в Kinesis, използвайки kinesis firehose plugin:

@type kinesis_firehose
    region region
    delivery_stream_name 
    aws_key_id 
    aws_sec_key

Athena

Ако сте настроили всичко правилно, след известно време (първоначално Kinesis записва получените данни на всеки 10 минути) трябва да видите файловете на логовете в S3. В менюто „monitoring“ на Kinesis Firehose можете да видите колко данни са записани в S3, както и грешки. Не забравяйте да предоставите достъп за запис в S3 кофа за ролята на Kinesis. Ако Kinesis не успее да парсне нещо, той ще запише грешките в същата кофа.

Сега можете да видите данните в Athena. Нека намерим новите заявки, по които сме получили грешки:

SELECT * FROM "db_name"."table_name" WHERE status > 499 ORDER BY created_at DESC limit 10;

Сканиране на всички записи при всяка заявка

Сега нашите логове са обработени и съхранени в S3 в ORC, компресирани и готови за анализ. Kinesis Firehose дори е организирал файловете по директории на всеки час. Въпреки това, докато таблицата не е разделена на партиции, Athena ще зарежда данните за всеки времеви период при всяка заявка, с изключение на редки случаи. Това е голям проблем по две причини:

• Обемът на данните постоянно нараства, което забавя заявките;
• Сметката за Athena се изчислява в зависимост от обема на сканираните данни, с минимум 10 MB за всяка заявка.

За да го поправим, използваме AWS Glue Crawler, който сканира данните в S3 и записва информация за партициите в Glue Metastore. Това ще ни позволи да използваме партиции като филтър при запитвания в Athena, и тя ще сканира само директориите, посочени в запитването.

Настройваме Amazon Glue Crawler

Amazon Glue Crawler сканира всички данни в S3 кофа и създава таблици с партиции. Създайте Glue Crawler от конзолата на AWS Glue и добавете кофата, в която съхранявате данни. Можете да използвате един краулер за няколко кофи, в този случай той ще създаде таблици в указаната база данни с имена, съвпадащи с имената на кофите. Ако планирате да използвате тези данни постоянно, не забравяйте да настроите разписание за стартиране на Crawler в съответствие с нуждите си. Използваме един Crawler за всички таблици, който стартира на всеки час.

Партиционирани таблици

След първото стартиране на краулера в базата данни, посочена в настройките, трябва да се появят таблици за всяка сканирана кофа. Отворете конзолата на Athena и намерете таблица с логовете на Nginx. Нека опитаме да прочетем нещо:

SELECT * FROM "default"."part_demo_kinesis_bucket"
WHERE(
  partition_0 = '2019' AND
  partition_1 = '04' AND
  partition_2 = '08' AND
  partition_3 = '06'
  );

Това запитване ще избере всички записи, получени от 6 до 7 сутринта на 8 април 2019 година. Но колко по-ефективно е това в сравнение с просто четене от непартиционирана таблица? Нека разберем и да изберем същите записи, филтрирайки ги по времеви штамп:

3.59 секунди и 244.34 мегабайта данни на датасет, в който има само седмица логове. Нека опитаме филтър по партиции:

Малко по-бързо, но най-важното — само 1.23 мегабайта данни! Това би било много по-евтино, ако не беше минималният лимит от 10 мегабайта за заявка в ценообразуването. Но все още е много по-добре, а при големи датасети разликата ще бъде още по-впечатляваща.

Собираем дэшборд с помощью Cube.js

За да съберем табло, използваме аналитичната рамка Cube.js. Тя предлага доста функции, но нас ни интересуват две: автоматично използване на филтри по партиции и предварителна агрегация на данни. Тя използва схема на данни data schema, написана на Javascript, за да генерира SQL и да изпълни запитване към базата данни. От нас се изисква само да посочим как да използваме филтера по партиции в схемата на данни.

Създаваме ново приложение Cube.js. Тъй като вече използваме AWS стек, е логично да използваме Lambda за разполагане. Можете да използвате express шаблон за генериране, ако планирате да хоствате Cube.js бекенда в Heroku или Docker. В документацията са описани и други методи за хостинг.

$ npm install -g cubejs-cli
$ cubejs create nginx-log-analytics -t serverless -d athena

За настройка на достъпа до базата данни в cube.js се използват променливи на средата. Генераторът ще създаде файл .env, в който можете да посочите вашите ключове за Athena.

Сега ще ни е необходима схема на данни, в която ще посочим как точно се съ хранят нашите логове. Там също може да се определи как да се изчисляват метриките за дашбордовете.

В директорията schema, създайте файл Logs.js. Ето пример на данна модел за nginx:

Код на модела

const partitionFilter = (from, to) => `
    date(from_iso8601_timestamp(${from})) = date_parse(partition_0 || partition_1 || partition_2, '%Y%m%d')
    `

cube(`Logs`, {
  sql: `
  select * from part_demo_kinesis_bucket
  WHERE ${FILTER_PARAMS.Logs.createdAt.filter(partitionFilter)}
  `,

  measures: {
    count: {
      type: `count`,
    },

    errorCount: {
      type: `count`,
      filters: [
        { sql: `${CUBE.isError} = 'Yes'` }
      ]
    },

    errorRate: {
      type: `number`,
      sql: `100.0 * ${errorCount} / ${count}`,
      format: `percent`
    }
  },

  dimensions: {
    status: {
      sql: `status`,
      type: `number`
    },

    isError: {
      type: `string`,
      case: {
        when: [{
          sql: `${CUBE}.status >= 400`, label: `Yes`
        }],
        else: { label: `No` }
      }
    },

    createdAt: {
      sql: `from_unixtime(created_at)`,
      type: `time`
    }
  }
});

Тук използваме променливата FILTER_PARAMS, за да генерираме SQL заявка с филтър по партиции.

Също така задаваме метриките и параметрите, които искаме да покажем на дашборда, и определяме пре-агрегациите. Cube.js ще създаде допълнителни таблици с пре-агрегирани данни и ще актуализира автоматично данните при постъпването им. Това позволява не само да се ускорят заявките, но и да се намалят разходите за използване на Athena.

Нека добавим тази информация в файла на схемата на данни:

preAggregations: {
  main: {
    type: `rollup`,
    measureReferences: [count, errorCount],
    dimensionReferences: [isError, status],
    timeDimensionReference: createdAt,
    granularity: `day`,
    partitionGranularity: `month`,
    refreshKey: {
      sql: FILTER_PARAMS.Logs.createdAt.filter((from, to) => 
        `select
           CASE WHEN from_iso8601_timestamp(${to}) + interval '3' day > now()
           THEN date_trunc('hour', now()) END`
      )
    }
  }
}

Упосочаваме в този модел, че е необходимо да се пре-агрегира информацията за всички използвани метрики, и да се използва партициониране по месеци. Партициониране на пре-агрегации може значително да ускори събирането и обновлението на данни.

Сега можем да съберем дашборд!

Бекендът на Cube.js предоставя REST API и набор клиентски библиотеки за популярни фронтенд фреймворкове. Ще използваме React версията на клиента за изграждане на дашборда. Cube.js предоставя само данни, така че ще ни е необходима библиотека за визуализации — на мен ми харесва recharts, но можете да използвате каквато и да е.

Сървърът Cube.js приема запитвания в JSON формат, в който са посочени необходимите метрики. Например, за да изчислим колко грешки е дал Nginx по дни, трябва да изпратим следното запитване:

{
  "measures": ["Logs.errorCount"],
  "timeDimensions": [
    {
      "dimension": "Logs.createdAt",
      "dateRange": ["2019-01-01", "2019-01-07"],
      "granularity": "day"
    }
  ]
}

Нека инсталираме клиента на Cube.js и библиотеката React компоненти чрез NPM:

$ npm i --save @cubejs-client/core @cubejs-client/react

Импортираме компонентите cubejs и QueryRenderer, за да извлечем данни, и създаваме дашборда:

Кодът на дашборда

import React from 'react';
import { LineChart, Line, XAxis, YAxis } from 'recharts';
import cubejs from '@cubejs-client/core';
import { QueryRenderer } from '@cubejs-client/react';

const cubejsApi = cubejs(
  'YOUR-CUBEJS-API-TOKEN',
  { apiUrl: 'http://localhost:4000/cubejs-api/v1' },
);

export default () => {
  return (
     {
        if (!resultSet) {
          return 'Loading...';
        }

        return (
          
            
            
            
          
        );
      }}
    />
  )
}

Източниците на дашборда са налични на CodeSandbox.

Източник: habr.com