Учебник: Автоматизация перекатывания с помощью ILM

При непрерывном индексировании временных документов в Elasticsearch, обычно используется поток данных, чтобы периодически переключаться на новый индекс. Это позволяет реализовать архитектуру «горячие-тёплые-холодные» для удовлетворения требований производительности для новых данных, контролировать затраты со временем, применять политики хранения и максимально эффективно использовать данные.

Для автоматизации переключения и управления потоком данных с помощью ILM, необходимо:

Политика жизненного цикла определяет фазы жизненного цикла индекса и действия, выполняемые в каждой фазе. Жизненный цикл может иметь до пяти фаз: hot, warm, cold, frozen и delete.

Например, вы можете определить timeseries_policy, имеющий две фазы:

Вы можете создать политику через Kibana или с помощью API создания или обновления политики. Чтобы создать политику из Kibana, откройте меню и перейдите к Управление стеком > Политики жизненного цикла индексов. Нажмите Создать политику.

resp = client.ilm.put_lifecycle(
    name="timeseries_policy",
    policy={
        "phases": {
            "hot": {
                "actions": {
                    "rollover": {
                        "max_primary_shard_size": "50GB",
                        "max_age": "30d"
                    }
                }
            },
            "delete": {
                "min_age": "90d",
                "actions": {
                    "delete": {}
                }
            }
        }
    },
)
print(resp)

response = client.ilm.put_lifecycle(
  policy: 'timeseries_policy',
  body: {
    policy: {
      phases: {
        hot: {
          actions: {
            rollover: {
              max_primary_shard_size: '50GB',
              max_age: '30d'
            }
          }
        },
        delete: {
          min_age: '90d',
          actions: {
            delete: {}
          }
        }
      }
    }
  }
)
puts response

const response = await client.ilm.putLifecycle({
  name: "timeseries_policy",
  policy: {
    phases: {
      hot: {
        actions: {
          rollover: {
            max_primary_shard_size: "50GB",
            max_age: "30d",
          },
        },
      },
      delete: {
        min_age: "90d",
        actions: {
          delete: {},
        },
      },
    },
  },
});
console.log(response);

PUT _ilm/policy/timeseries_policy
{
  "policy": {
    "phases": {
      "hot": {                                
        "actions": {
          "rollover": {
            "max_primary_shard_size": "50GB", 
            "max_age": "30d"
          }
        }
      },
      "delete": {
        "min_age": "90d",                     
        "actions": {
          "delete": {}                        
        }
      }
    }
  }
}

Для настройки потока данных сначала создайте шаблон индекса для указания политики жизненного цикла. Поскольку шаблон предназначен для потока данных, он также должен включать определение data_stream.

Например, вы можете создать timeseries_template для использования в будущем потоке данных с именем timeseries.

Чтобы включить ILM в управление потоком данных, шаблон настраивает одну настройку ILM:

Вы можете использовать мастер создания шаблона Kibana для добавления шаблона. В Kibana откройте меню и перейдите к Управление стеком > Управление индексами. На вкладке Шаблоны индексов нажмите Создать шаблон.

Этот мастер запускает API создания или обновления шаблона индекса для создания шаблона индекса с указанными параметрами.

resp = client.indices.put_index_template(
    name="timeseries_template",
    index_patterns=[
        "timeseries"
    ],
    data_stream={},
    template={
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 1,
            "index.lifecycle.name": "timeseries_policy"
        }
    },
)
print(resp)

response = client.indices.put_index_template(
  name: 'timeseries_template',
  body: {
    index_patterns: [
      'timeseries'
    ],
    data_stream: {},
    template: {
      settings: {
        number_of_shards: 1,
        number_of_replicas: 1,
        'index.lifecycle.name' => 'timeseries_policy'
      }
    }
  }
)
puts response

const response = await client.indices.putIndexTemplate({
  name: "timeseries_template",
  index_patterns: ["timeseries"],
  data_stream: {},
  template: {
    settings: {
      number_of_shards: 1,
      number_of_replicas: 1,
      "index.lifecycle.name": "timeseries_policy",
    },
  },
});
console.log(response);

PUT _index_template/timeseries_template
{
  "index_patterns": ["timeseries"],                   
  "data_stream": { },
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1,
      "index.lifecycle.name": "timeseries_policy"     
    }
  }
}

Для начала проиндексируйте документ в имя или шаблон подстановок, определенные в index_patterns шаблона индекса. Пока не существует существующий поток данных, индекс или псевдоним индекса с таким именем, запрос индекса автоматически создаст соответствующий поток данных с одним поддерживающим индексом. Elasticsearch автоматически индексирует документы запроса в этот поддерживающий индекс, который также действует как индекс записи потока.

Например, следующий запрос создаёт поток данных timeseries и первый поддерживающий индекс под названием .ds-timeseries-2099.03.08-000001.

resp = client.index(
    index="timeseries",
    document={
        "message": "logged the request",
        "@timestamp": "1591890611"
    },
)
print(resp)

response = client.index(
  index: 'timeseries',
  body: {
    message: 'logged the request',
    "@timestamp": '1591890611'
  }
)
puts response

const response = await client.index({
  index: "timeseries",
  document: {
    message: "logged the request",
    "@timestamp": "1591890611",
  },
});
console.log(response);

POST timeseries/_doc
{
  "message": "logged the request",
  "@timestamp": "1591890611"
}

При выполнении условия переключения в политике жизненного цикла действие rollover:

Этот процесс повторяется каждый раз, когда выполняется условие переключения. Вы можете искать во всех поддерживающих индексах потока данных, управляемых timeseries_policy, с именем потока данных timeseries. Операции записи должны отправляться в имя потока данных, которое направит их на текущий индекс записи. Операции чтения по потоку данных будут обрабатываться всеми его поддерживающими индексами.

Для получения информации о состоянии управляемых индексов используется API ILM explain. Это позволяет узнать, например:

Например, следующий запрос получает информацию о поддерживающих индексах потока данных timeseries:

resp = client.ilm.explain_lifecycle(
    index=".ds-timeseries-*",
)
print(resp)

response = client.ilm.explain_lifecycle(
  index: '.ds-timeseries-*'
)
puts response

const response = await client.ilm.explainLifecycle({
  index: ".ds-timeseries-*",
});
console.log(response);

GET .ds-timeseries-*/_ilm/explain

Следующий ответ показывает, что базовый индекс первого поколения потока данных ожидает действия hot фазы rollover. Он остаётся в этом состоянии, и ILM продолжает вызывать check-rollover-ready, пока не будут выполнены условия перехода.

{
  "indices": {
    ".ds-timeseries-2099.03.07-000001": {
      "index": ".ds-timeseries-2099.03.07-000001",
      "index_creation_date_millis": 1538475653281,
      "time_since_index_creation": "30s",        
      "managed": true,
      "policy": "timeseries_policy",             
      "lifecycle_date_millis": 1538475653281,
      "age": "30s",                              
      "phase": "hot",
      "phase_time_millis": 1538475653317,
      "action": "rollover",
      "action_time_millis": 1538475653317,
      "step": "check-rollover-ready",            
      "step_time_millis": 1538475653317,
      "phase_execution": {
        "policy": "timeseries_policy",
        "phase_definition": {                    
          "min_age": "0ms",
          "actions": {
            "rollover": {
              "max_primary_shard_size": "50gb",
              "max_age": "30d"
            }
          }
        },
        "version": 1,
        "modified_date_in_millis": 1539609701576
      }
    }
  }
}

Несмотря на то, что потоки данных являются удобным способом масштабирования и управления данными временных рядов, они разработаны для добавления данных только в конец. Мы признаём, что могут быть случаи, когда данные необходимо обновлять или удалять на месте, а потоки данных не поддерживают запросы на удаление и обновление напрямую, поэтому для работы с данными потока данных необходимо использовать API индексов напрямую. В этих случаях мы по-прежнему рекомендуем использовать поток данных.

Если вы часто отправляете несколько документов с помощью того же _id, ожидая, что последнее записанное значение будет перекрывать предыдущие, вы можете использовать псевдоним индекса вместо потока данных для управления индексами, содержащими данные временных рядов, и периодически переходить к новому индексу.

Для автоматизации перехода и управления индексами временных рядов с помощью ILM с использованием псевдонима индекса, вам необходимо:

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

Например, вы можете создать timeseries_template, который применяется к новым индексам, имена которых соответствуют шаблону индекса timeseries-*.

Для активации автоматического перехода шаблон настраивает два параметра ILM:

Вы можете использовать мастер Kibana по созданию шаблонов для добавления шаблона. Чтобы открыть мастер, откройте меню и перейдите к Управление стеком > Управление индексами. На вкладке Шаблоны индексов нажмите Создать шаблон.

Запрос на создание шаблона для примера шаблона выглядит следующим образом:

resp = client.indices.put_index_template(
    name="timeseries_template",
    index_patterns=[
        "timeseries-*"
    ],
    template={
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 1,
            "index.lifecycle.name": "timeseries_policy",
            "index.lifecycle.rollover_alias": "timeseries"
        }
    },
)
print(resp)

const response = await client.indices.putIndexTemplate({
  name: "timeseries_template",
  index_patterns: ["timeseries-*"],
  template: {
    settings: {
      number_of_shards: 1,
      number_of_replicas: 1,
      "index.lifecycle.name": "timeseries_policy",
      "index.lifecycle.rollover_alias": "timeseries",
    },
  },
});
console.log(response);

PUT _index_template/timeseries_template
{
  "index_patterns": ["timeseries-*"],                 
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1,
      "index.lifecycle.name": "timeseries_policy",      
      "index.lifecycle.rollover_alias": "timeseries"    
    }
  }
}

Для начала вам нужно инициализировать начальный индекс и назначить его индексом записи для псевдонима перехода, указанного в вашем шаблоне индекса. Имя этого индекса должно соответствовать шаблону индекса и заканчиваться числом. При переходе это значение увеличивается, чтобы сгенерировать имя для нового индекса.

Например, следующий запрос создаёт индекс под названием timeseries-000001 и делает его индексом записи для псевдонима timeseries.

resp = client.indices.create(
    index="timeseries-000001",
    aliases={
        "timeseries": {
            "is_write_index": True
        }
    },
)
print(resp)

const response = await client.indices.create({
  index: "timeseries-000001",
  aliases: {
    timeseries: {
      is_write_index: true,
    },
  },
});
console.log(response);

PUT timeseries-000001
{
  "aliases": {
    "timeseries": {
      "is_write_index": true
    }
  }
}

Когда выполняются условия перехода, действие rollover:

Этот процесс повторяется каждый раз, когда выполняются условия перехода. Вы можете искать во всех индексах, управляемых timeseries_policy с помощью псевдонима timeseries. Операции записи должны отправляться в псевдоним, который будет направлять их в текущий индекс записи.

Получение информации о состоянии управляемых индексов очень похоже на случай с потоками данных. См. раздел проверки прогресса потока данных для получения дополнительной информации. Единственное различие заключается в пространстве имен индексов, поэтому получение прогресса потребует следующего вызова API:

resp = client.ilm.explain_lifecycle(
    index="timeseries-*",
)
print(resp)

response = client.ilm.explain_lifecycle(
  index: 'timeseries-*'
)
puts response

const response = await client.ilm.explainLifecycle({
  index: "timeseries-*",
});
console.log(response);

GET timeseries-*/_ilm/explain