Simple Log Service:Flink SQL基于SPL实现弱结构化分析

{
  "Payload": "{\"lastNotified\": 1705030483, \"serverUri\": \"http://test.alert.com/alert-api/tasks\", \"jobID\": \"44d6ce47bb4995ef0c8052a9a30ed6d8\", \"alertName\": \"alert-12345678-123456\", \"project\": \"test-sls-project\", \"projectId\": 123, \"aliuid\": \"1234567890\", \"alertDisplayName\": \"\\u6d4b\\u8bd5\\u963f\\u91cc\\u4e91\\u544a\\u8b66\", \"checkJobUri\": \"http://test.alert.com/alert-api/task_check\", \"schedule\": {\"timeZone\": \"\", \"delay\": 0, \"runImmediately\": false, \"type\": \"FixedRate\", \"interval\": \"1m\"}, \"jobRunID\": \"bf86aa5e67a6891d-61016da98c79b-5071a6b\", \"firedNotNotified\": 25161}",
  "TaskID": "bf86aa5e67a6891d-61016da98c79b-5071a6b-334f81a-5c38aaa1-9354-43ec-8369-4f41a7c23887",
  "TaskType": "ALERT",
  "__source__": "11.199.XXX.XXX",
  "__tag__:__hostname__": "iabcde12345.cloud.abc121",
  "__tag__:__path__": "/var/log/service_a.LOG",
  "caller": "executor/pool.go:64",
  "error": "CouldNotExecuteQuery : {\n    \"httpCode\": 404,\n    \"errorCode\": \"LogStoreNotExist\",\n    \"errorMessage\": \"logstore k8s-event does not exist\",\n    \"requestID\": \"65B7C10AB43D9895A8C3DB6A\"\n}",
  "requestURL": "/apis/autoscaling/v2beta1/namespaces/python-etl/horizontalpodautoscalers/cn-shenzhen-56492-1234567890123?timeout=30s",
  "ts": "2024-01-29 22:57:13"
}

1. Prepare os dados no SLS

1. Certifique-se de ter ativado o Simple Log Service e criado um projeto e logstore.

2. Grave o trecho de log anterior no seu logstore de destino usando um SDK para simular dados de amostra.

   Após a gravação dos dados, acesse a aba Raw Logs no console do Simple Log Service para visualizar os dados simulados. O conteúdo do log inclui campos como TaskType (com o valor ALERT), Payload (que contém alertName, project e aliuid) e informações de erro (código de erro LogStoreNotExist, mensagem logstore k8s-event does not exist e status HTTP 404).

3. No logstore, escreva uma instrução de pipeline SPL do SLS e visualize os resultados.

   Depois de executar a instrução SPL, a aba Raw Logs exibe os registros de log analisados, que incluem campos estruturados como errorCode, errorMessage, fileName, fileNo, httpCode, project, requestID, scheduleType e serviceHost.

   A instrução de consulta SPL é a seguinte. A sintaxe de pipeline SPL usa o delimitador pipe (|) para separar comandos. Insira um comando por vez para ver o resultado imediato e adicione mais pipes para construir iterativamente a consulta final. Para obter mais informações, consulte Sintaxe de consulta de varredura.

   * | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller 
    | parse-json Payload 
    | project-away Payload 
    | parse-regexp error, 'CouldNotExecuteQuery : ({[\w":\s,\-}]+)' as errorJson 
    | parse-json errorJson 
    | parse-regexp "__tag__:__path__", '\/var\/log\/([\w\_]+).LOG' as serviceName 
    | parse-regexp caller, '\w+/([\w\.]+):(\d+)' as fileName, fileNo 
    | project-rename serviceHost="__tag__:__hostname__" 
    | extend scheduleType = json_extract_scalar(schedule, '$.type') 
    | project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType, project

   Explicação da sintaxe:

   • Linha 1: O comando project retém os campos Payload, error, __tag__:__path__ e caller para análise e descarta todos os outros campos.

   • Linha 2: O comando parse-json expande a string Payload em um objeto JSON. Seus campos de nível superior, como lastNotified, serviceUri e jobID, são adicionados ao resultado.

   • Linha 3: O comando project-away remove o campo Payload original.

   • Linha 4: O comando parse-regexp utiliza uma expressão regular para extrair a parte JSON do campo error e a atribui a um novo campo chamado errorJson.

   • Linha 5: O comando parse-json expande o campo errorJson, extraindo httpCode, errorCode e errorMessage.

   • Linha 6: O comando parse-regexp usa uma expressão regular para extrair o nome do arquivo de __tag__:__path__ e o nomeia como serviceName.

   • Linha 7: O comando parse-regexp aplica uma expressão regular para extrair o nome do arquivo e o número da linha do campo caller, atribuindo-os aos campos fileName e fileNo.

   • Linha 8: O comando project-rename altera o nome do campo __tag__:__hostname__ para serviceHost.

   • Linha 9: O comando extend emprega a função json_extract_scalar para extrair o campo type do objeto schedule e o nomeia como scheduleType.

   • Linha 10: O comando project mantém apenas os campos finais necessários, incluindo o campo project que foi extraído de Payload.

2. Crie uma tarefa SQL

1. Faça login no console do Realtime Compute for Apache Flink e clique no workspace de destino.

2. No painel de navegação à esquerda, escolha Data Development > ETL.

3. Clique em Create. Na caixa de diálogo New Draft, escolha SQL Scripts > Blank Stream Draft e clique em Next.

4. No editor de rascunho, insira a seguinte instrução para criar uma tabela temporária.

   CREATE TEMPORARY TABLE sls_input_complex (
     errorCode STRING,
     errorMessage STRING,
     fileName STRING,
     fileNo STRING,
     httpCode STRING,
     requestID STRING,
     scheduleType STRING,
     serviceHost STRING,
     project STRING,
     proctime as PROCTIME()
   ) WITH (
     'connector' = 'sls',
     'endpoint' ='cn-beijing-intranet.log.aliyuncs.com',
     'accessId' = '${yourAccessKeyID}',
     'accessKey' = '${yourAccessKeySecret}',
     'starttime' = '2024-02-01 10:30:00',
     'project' ='${project}',
     'logstore' ='${logtore}',
     'query' = '* | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller | parse-json Payload | project-away Payload | parse-regexp error, ''CouldNotExecuteQuery : ({[\w":\s,\-}]+)'' as errorJson | parse-json errorJson | parse-regexp "__tag__:__path__", ''\/var\/log\/([\w\_]+).LOG'' as serviceName | parse-regexp caller, ''\w+/([\w\.]+):(\d+)'' as fileName, fileNo | project-rename serviceHost="__tag__:__hostname__" | extend scheduleType = json_extract_scalar(schedule, ''$.type'') | project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType,project'
     );

   A tabela a seguir descreve os parâmetros na cláusula WITH. Substitua os valores de exemplo pelos seus valores reais.

   Parâmetro	Descrição	Exemplo
   connector	Tipo de conector. Conectores suportados.	sls
   endpoint	Endpoint interno do SLS. Endpoints.	cn-hangzhou-intranet.log.aliyuncs.com
   accessId	Seu AccessKey ID. Criar um AccessKey.	LTAI
   accessKey	O AccessKey Secret. Criar um AccessKey.	yourAccessKeySecret
   starttime	Hora inicial para consumo de logs.	2025-02-19 00:00:00
   project	Nome do projeto SLS.	test-project
   logstore	Nome do Logstore SLS.	clb-access-log
   query	Instrução SPL. Escape literais de string com aspas simples duplicadas ('') no Flink SQL.	*	where slbid = ''slb-01''

5. Selecione a instrução SQL, clique com o botão direito e selecione Running para conectar-se ao Simple Log Service.

   CREATE TEMPORARY TABLE sls_input_complex (
     errorCode STRING,
     errorMessage STRING,
     fileName STRING,
     fileNo STRING,
     httpCode STRING,
     requestID STRING,
     scheduleType STRING,
     serviceHost STRING,
     project STRING,
     proctime as PROCTIME()
   ) WITH (
     'connector' = 'sls',
     'endpoint' = 'cn-hxxx',
     'accessId' = 'xxx',
     'accessKey' = 'xxx',
     'starttime' = 'xxx',
     'project' = 'xxx',
     'logstore' = 'clb7xxx',
     'query' = '* | prxxx", "__tag__:__hosxxx'
   );

3. Execute a consulta e visualize os resultados

1. No editor de tarefas, insira a seguinte instrução para consultar os dados:

   SELECT * FROM sls_input_complex;

2. Clique em Debug no canto superior direito. Na caixa de diálogo, selecione Create new session cluster na lista suspensa Session Cluster e configure conforme abaixo.

   Defina Name como demo-test, Deployment Target como default-queue, Status como RUNNING e Engine Version como vvr-8.0.11-flink-1.17, depois clique em Create Session Cluster.

3. Na caixa de diálogo de depuração, selecione o cluster de sessão recém-criado e clique em OK.

   Nota: A depuração com uma tabela de origem SLS avança o deslocamento do grupo de consumidores. Uma tarefa implantada retoma a partir desse novo deslocamento.

4. Na aba Results, observe que cada coluna na tabela corresponde a um campo processado pela consulta SPL.

   Após executar o Debug da consulta, a tabela de resultados mostra os campos analisados. Por exemplo, a coluna errorCode apresenta LogStoreNotExist, a coluna errorMessage exibe logstore k8s-event does not exist e a coluna httpCode mostra 404.