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. Préparer les données dans SLS

1. Assurez-vous d'avoir activé Simple Log Service et créé un projet et un logstore.

2. Écrivez l'extrait de journal précédent dans votre logstore cible à l'aide d'un SDK pour simuler des exemples de données.

   Une fois les données écrites, accédez à l'onglet Raw Logs dans la console Simple Log Service pour afficher les données simulées. Le contenu du journal inclut des champs tels que TaskType (avec la valeur ALERT), Payload (qui contient alertName, project et aliuid), et des informations d'erreur (code d'erreur LogStoreNotExist, message logstore k8s-event does not exist et statut HTTP 404).

3. Dans le logstore, écrivez une instruction de pipeline SLS SPL et prévisualisez les résultats.

   Après avoir exécuté l'instruction SPL, l'onglet Raw Logs affiche les enregistrements de journal analysés, qui incluent des champs structurés tels que errorCode, errorMessage, fileName, fileNo, httpCode, project, requestID, scheduleType et serviceHost.

   L'instruction de requête SPL est la suivante. La syntaxe du pipeline SPL utilise le délimiteur pipe (|) pour séparer les commandes. Vous pouvez saisir une commande à la fois pour voir le résultat immédiat, puis ajouter d'autres pipes pour construire itérativement la requête finale. Pour plus d'informations, consultez Syntaxe de requête d'analyse.

   * | 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

   Explication de la syntaxe :

   • Ligne 1 : La commande project conserve les champs Payload, error, __tag__:__path__ et caller pour l'analyse et ignore tous les autres champs.

   • Ligne 2 : La commande parse-json développe la chaîne Payload en un objet JSON. Ses champs de niveau supérieur, tels que lastNotified, serviceUri et jobID, sont ajoutés au résultat.

   • Ligne 3 : La commande project-away supprime le champ Payload d'origine.

   • Ligne 4 : La commande parse-regexp utilise une expression régulière pour extraire la partie JSON du champ error et l'assigne à un nouveau champ nommé errorJson.

   • Ligne 5 : La commande parse-json développe le champ errorJson, en extrayant httpCode, errorCode et errorMessage.

   • Ligne 6 : La commande parse-regexp utilise une expression régulière pour extraire le nom de fichier de __tag__:__path__ et le nomme serviceName.

   • Ligne 7 : La commande parse-regexp utilise une expression régulière pour extraire le nom de fichier et le numéro de ligne du champ caller et les assigne aux champs fileName et fileNo.

   • Ligne 8 : La commande project-rename renomme le champ __tag__:__hostname__ en serviceHost.

   • Ligne 9 : La commande extend utilise la fonction json_extract_scalar pour extraire le champ type de l'objet schedule et le nomme scheduleType.

   • Ligne 10 : La commande project ne conserve que les champs finaux requis, y compris le champ project qui a été extrait de Payload.

2. Créer une tâche SQL

1. Connectez-vous à la console Realtime Compute for Apache Flink et cliquez sur l'espace de travail cible.

2. Dans le volet de navigation de gauche, choisissez Data Development > ETL.

3. Cliquez sur Create. Dans la boîte de dialogue New Draft, choisissez SQL Scripts > Blank Stream Draft, puis cliquez sur Next.

4. Dans l'éditeur de brouillon, saisissez l'instruction suivante pour créer une table temporaire.

   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'
     );

   Le tableau suivant décrit les paramètres de la clause WITH. Remplacez les valeurs d'exemple par vos valeurs réelles.

   Parameter	Description	Example
   connector	Le type de connecteur. Connecteurs pris en charge.	sls
   endpoint	L'endpoint interne SLS. Endpoints.	cn-hangzhou-intranet.log.aliyuncs.com
   accessId	Votre AccessKey ID. Créer une AccessKey.	LTAI
   accessKey	L'AccessKey Secret. Créer une AccessKey.	yourAccessKeySecret
   starttime	Heure de début pour la consommation des journaux.	2025-02-19 00:00:00
   project	Le nom du projet SLS.	test-project
   logstore	Le nom du Logstore SLS.	clb-access-log
   query	L'instruction SPL. Échappez les littéraux de chaîne avec des guillemets simples doublés ('') dans Flink SQL.	*	where slbid = ''slb-01''

5. Sélectionnez l'instruction SQL, faites un clic droit et sélectionnez Running pour vous connecter à 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. Exécuter la requête et afficher les résultats

1. Dans l'éditeur de tâches, saisissez l'instruction suivante pour interroger les données :

   SELECT * FROM sls_input_complex;

2. Cliquez sur Debug dans le coin supérieur droit. Dans la boîte de dialogue, sélectionnez Create new session cluster dans la liste déroulante Session Cluster et configurez comme suit.

   Définissez Name sur demo-test, Deployment Target sur default-queue, Status sur RUNNING et Engine Version sur vvr-8.0.11-flink-1.17, puis cliquez sur Create Session Cluster.

3. Dans la boîte de dialogue de débogage, sélectionnez le cluster de session que vous venez de créer, puis cliquez sur OK.

   Remarque : Le débogage avec une table source SLS avance le décalage du groupe de consommateurs. Une tâche déployée reprend à partir de ce nouveau décalage.

4. Dans l'onglet Results, vous pouvez voir que chaque colonne de la table correspond à un champ traité par la requête SPL.

   Après avoir Debug la requête, la table de résultats affiche les champs analysés. Par exemple, la colonne errorCode est LogStoreNotExist, la colonne errorMessage est logstore k8s-event does not exist et la colonne httpCode est 404.