Elasticsearch:Démarrage rapide : Gérer les données de séries temporelles Elasticsearch avec TimeStream

Créer un index de série temporelle

Créez un index de série temporelle nommé test_stream :

PUT _time_stream/test_stream

Cette commande crée un flux de données Elasticsearch (et non un index autonome), ainsi qu'un modèle d'index nommé .timestream_test_stream qui préconfigure les paramètres suivants :

Le modèle d'index préconfigure également les mappages pour deux catégories de champs :

• Champs de dimension (labels.*) : Mappés comme keyword avec time_series_dimension: true. Tous les champs de dimension sont combinés dans le champ interne _tsid, qui identifie de manière unique chaque série temporelle.

• Champs métriques (metrics.*) : Mappés comme double ou long avec "index": false. Seuls les doc values sont stockés, sans données d'index inversé.

Consultez la configuration complète de test_stream :

GET _time_stream/test_stream

Personnaliser le modèle d'index

Transmettez un corps template lors de la création de l'index pour remplacer les paramètres par défaut. Les exemples suivants illustrent des personnalisations courantes :

• Définissez le nombre de shards primaires :

  PUT _time_stream/test_stream
  {
    "template": {
      "settings": {
        "index": {
          "number_of_shards": "2"
        }
      }
    }
  }

• Personnalisez le modèle de données (conventions de nommage des champs) :

  PUT _time_stream/test_stream
  {
    "template": {
      "settings": {
        "index": {
          "number_of_shards": "2"
        }
      }
    },
    "time_stream": {
      "labels_fields": ["labels_*"],
      "metrics_fields": ["metrics_*"]
    }
  }

Mettre à jour un index de série temporelle

Mettez à jour le nombre de shards primaires pour test_stream :

POST _time_stream/test_stream/_update
{
  "template": {
    "settings": {
      "index": {
        "number_of_shards": "4"
      }
    }
  }
}

Après la mise à jour, les nouveaux paramètres ne s'appliquent pas à l'index sous-jacent actuel. Effectuez un basculement (rollover) de l'index pour les appliquer :

POST test_stream/_rollover

Un nouvel index sous-jacent est créé avec les paramètres mis à jour. L'index sous-jacent d'origine conserve ses paramètres précédents.

Supprimer un index de série temporelle

Supprimez test_stream et toutes ses données :

DELETE _time_stream/test_stream

Écrire des données

Utilisez l'API bulk ou index pour écrire des documents. Chaque document doit inclure un champ @timestamp correspondant à l'heure de la mesure.

POST test_stream/_doc
{
  "@timestamp": 1630465208722,
  "metrics": {
    "cpu.idle": 79.67298116109929,
    "disk_ioutil": 17.630910821570456,
    "mem.free": 75.79973639970004
  },
  "labels": {
    "disk_type": "disk_type2",
    "namespace": "namespaces1",
    "clusterId": "clusterId3",
    "nodeId": "nodeId5"
  }
}

Fonctionnement du routage basé sur le temps

Chaque index sous-jacent du flux de données possède une plage temporelle définie par index.time_series.start_time et index.time_series.end_time. Un document est écrit dans l'index sous-jacent dont la plage temporelle contient la valeur @timestamp du document.

Le flux de données gère cette plage temporelle automatiquement. Après un basculement, le nouvel index sous-jacent prend le relais à partir de l'heure de fin de l'index précédent. Ainsi, l'ensemble des index sous-jacents couvre une plage temporelle continue sans interruption.

Interroger des données

Recherchez tous les documents dans test_stream :

GET test_stream/_search

Consultez les détails de l'index :

GET _cat/indices/test_stream?v&s=i

Consulter les statistiques de l'index

Obtenez les statistiques pour test_stream, y compris le nombre de séries temporelles suivies par shard :

GET _time_stream/test_stream/_stats

Exemple de réponse :

{
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "failed" : 0
  },
  "time_stream_count" : 1,
  "indices_count" : 1,
  "total_store_size_bytes" : 19132,
  "time_streams" : [
    {
      "time_stream" : "test_stream",
      "indices_count" : 1,
      "store_size_bytes" : 19132,
      "tsid_count" : 2
    }
  ]
}

La métrique time_stream_count comptabilise les séries temporelles uniques par shard primaire en lisant les doc values du champ _tsid. Cette opération étant relativement coûteuse, réduisez son coût comme suit :

• Pour les index en lecture seule, configurez une politique de mise en cache afin que le comptage ne soit effectué qu'une seule fois.

• Pour les index actifs, le cache s'actualise toutes les 5 minutes par défaut. Ajustez cet intervalle avec le paramètre index.time_series.stats.refresh_interval. La valeur minimale est de 1 minute.

Configurer une source de données Prometheus

Option 1 : Console Grafana

Dans la console Grafana, ajoutez une source de données Prometheus et définissez l'URL sur /_time_stream/prom/test_stream. L'index de série temporelle est alors directement disponible en tant que source de données Prometheus.

Autres éléments de configuration importants :

• Définissez Network Type sur Public Network.

• Définissez Access sur Server (default).

• Activez Basic auth, définissez User sur elastic, et saisissez le mot de passe correspondant dans Password.

Option 2 : Préfixes et suffixes de champs personnalisés

Lors d'une requête via l'API Prometheus, le plug-in retire par défaut le préfixe metrics. des champs métriques et le préfixe labels. des champs de dimension.

Si vous avez créé l'index avec un modèle de données personnalisé (préfixes de champs non par défaut), configurez explicitement les mappages de préfixes et de suffixes :

PUT _time_stream/{name}
{
  "time_stream": {
    "labels_fields": "@labels.*_l",
    "metrics_fields": "@metrics.*_m",
    "label_prefix": "@labels.",
    "label_suffix": "_l",
    "metric_prefix": "@metrics.",
    "metric_suffix": "_m"
  }
}

Interroger les métadonnées

Consultez tous les champs métriques dans test_stream :

GET /_time_stream/prom/test_stream/metadata

Exemple de réponse :

{
  "status" : "success",
  "data" : {
    "cpu.idle" : [
      {
        "type" : "gauge",
        "help" : "",
        "unit" : ""
      }
    ],
    "disk_ioutil" : [
      {
        "type" : "gauge",
        "help" : "",
        "unit" : ""
      }
    ],
    "mem.free" : [
      {
        "type" : "gauge",
        "help" : "",
        "unit" : ""
      }
    ]
  }
}

Consultez tous les champs de dimension dans test_stream :

GET /_time_stream/prom/test_stream/labels

Exemple de réponse :

{
  "status" : "success",
  "data" : [
    "__name__",
    "clusterId",
    "disk_type",
    "namespace",
    "nodeId"
  ]
}

Consultez toutes les valeurs d'un champ de dimension spécifique :

GET /_time_stream/prom/test_stream/label/clusterId/values

Exemple de réponse :

{
  "status" : "success",
  "data" : [
    "clusterId1",
    "clusterId3"
  ]
}

Consultez toutes les séries temporelles pour la métrique cpu.idle :

GET /_time_stream/prom/test_stream/series?match[]=cpu.idle

Exemple de réponse :

{
  "status" : "success",
  "data" : [
    {
      "__name__" : "cpu.idle",
      "disk_type" : "disk_type1",
      "namespace" : "namespaces2",
      "clusterId" : "clusterId1",
      "nodeId" : "nodeId2"
    },
    {
      "__name__" : "cpu.idle",
      "disk_type" : "disk_type1",
      "namespace" : "namespaces2",
      "clusterId" : "clusterId1",
      "nodeId" : "nodeId5"
    },
    {
      "__name__" : "cpu.idle",
      "disk_type" : "disk_type2",
      "namespace" : "namespaces1",
      "clusterId" : "clusterId3",
      "nodeId" : "nodeId5"
    }
  ]
}

Interroger des données avec PromQL

Utilisez les API de requête instantanée et de requête par plage de Prometheus pour exécuter des requêtes PromQL sur votre index de série temporelle. Pour plus de détails sur la syntaxe PromQL prise en charge, consultez Prise en charge de PromQL par aliyun-timestream.

Requête instantanée

GET /_time_stream/prom/test_stream/query?query=cpu.idle&time=1655769837

Le paramètre time est exprimé en secondes Unix. S'il est omis, la requête renvoie les données des 5 dernières minutes.

Exemple de réponse :

{
  "status" : "success",
  "data" : {
    "resultType" : "vector",
    "result" : [
      {
        "metric" : {
          "__name__" : "cpu.idle",
          "clusterId" : "clusterId1",
          "disk_type" : "disk_type1",
          "namespace" : "namespaces2",
          "nodeId" : "nodeId2"
        },
        "value" : [
          1655769837,
          "79.672981161"
        ]
      },
      {
        "metric" : {
          "__name__" : "cpu.idle",
          "clusterId" : "clusterId1",
          "disk_type" : "disk_type1",
          "namespace" : "namespaces2",
          "nodeId" : "nodeId5"
        },
        "value" : [
          1655769837,
          "79.672981161"
        ]
      },
      {
        "metric" : {
          "__name__" : "cpu.idle",
          "clusterId" : "clusterId3",
          "disk_type" : "disk_type2",
          "namespace" : "namespaces1",
          "nodeId" : "nodeId5"
        },
        "value" : [
          1655769837,
          "79.672981161"
        ]
      }
    ]
  }
}

Requête par plage

GET /_time_stream/prom/test_stream/query_range?query=cpu.idle&start=1655769800&end=16557699860&step=1m

Exemple de réponse :

{
  "status" : "success",
  "data" : {
    "resultType" : "matrix",
    "result" : [
      {
        "metric" : {
          "__name__" : "cpu.idle",
          "clusterId" : "clusterId1",
          "disk_type" : "disk_type1",
          "namespace" : "namespaces2",
          "nodeId" : "nodeId2"
        },
        "value" : [
          [
            1655769860,
            "79.672981161"
          ]
        ]
      },
      {
        "metric" : {
          "__name__" : "cpu.idle",
          "clusterId" : "clusterId1",
          "disk_type" : "disk_type1",
          "namespace" : "namespaces2",
          "nodeId" : "nodeId5"
        },
        "value" : [
          [
            1655769860,
            "79.672981161"
          ]
        ]
      },
      {
        "metric" : {
          "__name__" : "cpu.idle",
          "clusterId" : "clusterId3",
          "disk_type" : "disk_type2",
          "namespace" : "namespaces1",
          "nodeId" : "nodeId5"
        },
        "value" : [
          [
            1655769860,
            "79.672981161"
          ]
        ]
      }
    ]
  }
}

PUT _time_stream/test_stream
{
  "time_stream": {
    "downsample": [
      {
        "interval": "1m"
      },
      {
        "interval": "10m"
      },
      {
        "interval": "60m"
      }
    ]
  }
}

GET test_stream/_search?size=0&request_cache=false
{
  "aggs": {
    "1": {
      "terms": {
        "field": "labels.disk_type",
        "size": 10
      },
      "aggs": {
        "2": {
          "date_histogram": {
            "field": "@timestamp",
            "fixed_interval": "120m"
          }
        }
      }
    }
  }
}

{
  "took" : 15,
  "timed_out" : false,
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "1" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "disk_type2",
          "doc_count" : 9,
          "2" : {
            "buckets" : [
              {
                "key_as_string" : "2022-06-20T06:00:00.000Z",
                "key" : 1655704800000,
                "doc_count" : 9
              }
            ]
          }
        }
      ]
    }
  }
}

Tester le downsampling avec un index de démonstration

La procédure suivante illustre le cycle de vie complet du downsampling à l'aide d'une plage temporelle contrôlée manuellement. Elle est utile pour les tests et la validation. En production, le plug-in déclenche le downsampling automatiquement après le basculement.

1. Créez un index avec une plage temporelle fixe et une configuration de downsampling. Les valeurs start_time et end_time simulent une fenêtre temporelle passée afin que la condition de déclenchement du downsampling (l'heure actuelle doit dépasser d'au moins deux heures end_time) soit immédiatement satisfaite.

   Important

   Après cette étape, vérifiez end_time en exécutant GET {index}/_settings. Le système ajuste end_time toutes les 5 minutes par défaut. Passez à l'étape suivante avant que end_time ne soit mis à jour.

   PUT _time_stream/test_stream
   {
     "template": {
       "settings": {
         "index.time_series.start_time": "2022-06-20T00:00:00.000Z",
         "index.time_series.end_time": "2022-06-21T00:00:00.000Z"
       }
     },
     "time_stream": {
       "downsample": [
         {
           "interval": "1m"
         },
         {
           "interval": "10m"
         },
         {
           "interval": "60m"
         }
       ]
     }
   }

2. Écrivez un document dont le champ @timestamp se situe dans la plage [start_time, end_time) :

   POST test_stream/_doc
   {
     "@timestamp": 1655706106000,
     "metrics": {
       "cpu.idle": 79.67298116109929,
       "disk_ioutil": 17.630910821570456,
       "mem.free": 75.79973639970004
     },
     "labels": {
       "disk_type": "disk_type2",
       "namespace": "namespaces1",
       "clusterId": "clusterId3",
       "nodeId": "nodeId5"
     }
   }

3. Retirez start_time et end_time de l'index, tout en conservant la configuration de downsampling :

   POST _time_stream/test_stream/_update
   {
     "time_stream": {
       "downsample": [
         {
           "interval": "1m"
         },
         {
           "interval": "10m"
         },
         {
           "interval": "60m"
         }
       ]
     }
   }

4. Effectuez le basculement de l'index :

   POST test_stream/_rollover

5. Une fois le basculement terminé, consultez les index de downsampling générés :

   GET _cat/indices/test_stream?v&s=i

   Résultat attendu :

   health status index                                          uuid                   pri rep docs.count docs.deleted store.size pri.store.size
   green  open   .ds-test_stream-2022.06.21-000001              vhEwKIlwSGO3ax4RKn****   1   1          9            0     18.5kb         12.1kb
   green  open   .ds-test_stream-2022.06.21-000001_interval_10m r9Tsj0v-SyWJDc64oC****   1   1          1            0     15.8kb          7.9kb
   green  open   .ds-test_stream-2022.06.21-000001_interval_1h  cKsAlMK-T2-luefNAF****   1   1          1            0     15.8kb          7.9kb
   green  open   .ds-test_stream-2022.06.21-000001_interval_1m  L6ocasDFTz-c89KjND****   1   1          1            0     15.8kb          7.9kb
   green  open   .ds-test_stream-2022.06.21-000002              42vlHEFFQrmMAdNdCz****   1   1          0            0       452b           226b

   Les trois index de downsampling (_interval_1m, _interval_10m, _interval_1h) sont créés parallèlement au nouvel index sous-jacent (000002).