---
type: api
title: Create or update a Logstash pipeline
description: Create a pipeline that is used for Logstash Central Management.
url: https://www.elastic.co/elastic/docs-builder/docs/4371/api/doc/elasticsearch-serverless/operation/operation-logstash-put-pipeline
resource: https://www.elastic.co/elastic/docs-builder/docs/4371/api/doc/elasticsearch-serverless/operation/operation-logstash-put-pipeline
products:
  - Elasticsearch Serverless
---

# Create or update a Logstash pipeline

Availability:

## Paths

- `PUT` `/_logstash/pipeline/{id}`

- [Api key auth](/elastic/docs-builder/docs/4371/api/doc/elasticsearch-serverless/authentication#apikeyauth)

#### Path Parameters

- `id` (string Id)
  An identifier for the pipeline.
Pipeline IDs must begin with a letter or underscore and contain only letters, underscores, dashes, hyphens and numbers.

## Description

Create a pipeline that is used for Logstash Central Management.
If the specified pipeline exists, it is replaced.

## Required authorization

* Cluster privileges: `manage_logstash_pipelines`

[Centralized pipeline management](https://www.elastic.co/docs/reference/logstash/logstash-centralized-pipeline-management)

## Request Body

`application/json`

- `description` (string) — required
  A description of the pipeline.
This description is not used by Elasticsearch or Logstash.
- `last_modified` (DateTime) — required
  The date the pipeline was last updated.
It must be in the `yyyy-MM-dd'T'HH:mm:ss.SSSZZ` strict_date_time format.
- `pipeline` (string) — required
  The configuration for the pipeline.
- `pipeline_metadata` ({} PipelineMetadata) — required
  Optional metadata about the pipeline, which can have any contents.
This metadata is not generated or used by Elasticsearch or Logstash.
  - `type` (string) — required
  - `version` (string) — required
- `pipeline_settings` ({} PipelineSettings) — required
  Settings for the pipeline.
It supports only flat keys in dot notation.
  - `pipeline.workers` (number) — required
    The number of workers that will, in parallel, execute the filter and output stages of the pipeline.
  - `pipeline.batch.size` (number) — required
    The maximum number of events an individual worker thread will collect from inputs before attempting to execute its filters and outputs.
  - `pipeline.batch.delay` (number) — required
    When creating pipeline event batches, how long in milliseconds to wait for each event before dispatching an undersized batch to pipeline workers.
  - `queue.type` (string) — required
    The internal queuing model to use for event buffering.
  - `queue.max_bytes` (string) — required
    The total capacity of the queue (`queue.type: persisted`) in number of bytes.
  - `queue.checkpoint.writes` (number) — required
    The maximum number of written events before forcing a checkpoint when persistent queues are enabled (`queue.type: persisted`).
- `username` (string) — required
  The user who last updated the pipeline.

## Response

## Code Examples

#### Console

```console
PUT _logstash/pipeline/my_pipeline
{
  "description": "Sample pipeline for illustration purposes",
  "last_modified": "2021-01-02T02:50:51.250Z",
  "pipeline_metadata": {
    "type": "logstash_pipeline",
    "version": 1
  },
  "username": "elastic",
  "pipeline": "input {}\\n filter { grok {} }\\n output {}",
  "pipeline_settings": {
    "pipeline.workers": 1,
    "pipeline.batch.size": 125,
    "pipeline.batch.delay": 50,
    "queue.type": "memory",
    "queue.max_bytes": "1gb",
    "queue.checkpoint.writes": 1024
  }
}
```

#### JavaScript

```javascript
const response = await client.logstash.putPipeline({
  id: "my_pipeline",
  pipeline: {
    description: "Sample pipeline for illustration purposes",
    last_modified: "2021-01-02T02:50:51.250Z",
    pipeline_metadata: {
      type: "logstash_pipeline",
      version: 1,
    },
    username: "elastic",
    pipeline: "input {}\\n filter { grok {} }\\n output {}",
    pipeline_settings: {
      "pipeline.workers": 1,
      "pipeline.batch.size": 125,
      "pipeline.batch.delay": 50,
      "queue.type": "memory",
      "queue.max_bytes": "1gb",
      "queue.checkpoint.writes": 1024,
    },
  },
});
```

#### Python

```python
resp = client.logstash.put_pipeline(
    id="my_pipeline",
    pipeline={
        "description": "Sample pipeline for illustration purposes",
        "last_modified": "2021-01-02T02:50:51.250Z",
        "pipeline_metadata": {
            "type": "logstash_pipeline",
            "version": 1
        },
        "username": "elastic",
        "pipeline": "input {}\\n filter { grok {} }\\n output {}",
        "pipeline_settings": {
            "pipeline.workers": 1,
            "pipeline.batch.size": 125,
            "pipeline.batch.delay": 50,
            "queue.type": "memory",
            "queue.max_bytes": "1gb",
            "queue.checkpoint.writes": 1024
        }
    },
)
```

#### curl

```curl
curl -X PUT "$ELASTICSEARCH_URL/_logstash/pipeline/my_pipeline" \
  -H "Authorization: ApiKey $ELASTIC_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "description": "Sample pipeline for illustration purposes",
    "last_modified": "2021-01-02T02:50:51.250Z",
    "pipeline_metadata": {
      "type": "logstash_pipeline",
      "version": 1
    },
    "username": "elastic",
    "pipeline": "input {}\\n filter { grok {} }\\n output {}",
    "pipeline_settings": {
      "pipeline.workers": 1,
      "pipeline.batch.size": 125,
      "pipeline.batch.delay": 50,
      "queue.type": "memory",
      "queue.max_bytes": "1gb",
      "queue.checkpoint.writes": 1024
    }
  }'
```

#### Java

```java
client.logstash().putPipeline(p -> p
    .id("my_pipeline")
    .pipeline(pi -> pi
        .description("Sample pipeline for illustration purposes")
        .lastModified(DateTime.of("2021-01-02T02:50:51.250Z"))
        .pipeline("input {}\n filter { grok {} }\n output {}")
        .pipelineMetadata(pip -> pip
            .type("logstash_pipeline")
            .version("1")
        )
        .pipelineSettings(pip -> pip
            .pipelineWorkers(1)
            .pipelineBatchSize(125)
            .pipelineBatchDelay(50)
            .queueType("memory")
            .queueMaxBytes("1gb")
            .queueCheckpointWrites(1024)
        )
        .username("elastic")
    )
);
```

#### PHP

```php
$resp = $client->logstash()->putPipeline([
    "id" => "my_pipeline",
    "body" => [
        "description" => "Sample pipeline for illustration purposes",
        "last_modified" => "2021-01-02T02:50:51.250Z",
        "pipeline_metadata" => [
            "type" => "logstash_pipeline",
            "version" => 1,
        ],
        "username" => "elastic",
        "pipeline" => "input {}\\n filter { grok {} }\\n output {}",
        "pipeline_settings" => [
            "pipeline.workers" => 1,
            "pipeline.batch.size" => 125,
            "pipeline.batch.delay" => 50,
            "queue.type" => "memory",
            "queue.max_bytes" => "1gb",
            "queue.checkpoint.writes" => 1024,
        ],
    ],
]);
```

#### Ruby

```ruby
response = client.logstash.put_pipeline(
  id: "my_pipeline",
  body: {
    "description": "Sample pipeline for illustration purposes",
    "last_modified": "2021-01-02T02:50:51.250Z",
    "pipeline_metadata": {
      "type": "logstash_pipeline",
      "version": 1
    },
    "username": "elastic",
    "pipeline": "input {}\\n filter { grok {} }\\n output {}",
    "pipeline_settings": {
      "pipeline.workers": 1,
      "pipeline.batch.size": 125,
      "pipeline.batch.delay": 50,
      "queue.type": "memory",
      "queue.max_bytes": "1gb",
      "queue.checkpoint.writes": 1024
    }
  }
)
```