Set up Logstash as an ingest method

Step 1 - Create a Logstash ingest token

Why? Ingest tokens authenticate and authorize Logstash instances to send data to your repository. They control which parsers can be used and what fields can be populated by Logstash.

Detailed steps:

Sign in to Falcon LogScale, and browse to your repository.
Click Settings, Ingest Tokens.
Click Add token.
Type in a descriptive name (for example, logstash-production-pipeline).
Set the appropriate permissions:
- Assign parser to allow Logstash to specify parsers based on log source type
- Assign fields to enable field creation from Logstash-processed data
Click Create token to save the token and securely store the generated string - you'll need this when configuring Logstash output.
Note your LogScale ingestion endpoint URL:
- For cloud deployments: Typically https://cloud.humio.com or your regional endpoint
- For on-premises deployments: Your self-hosted LogScale URL
- The full ingestion endpoint will be: https://cloud.humio.com/api/v1/ingest/humio-structured

Step 2 - Plan your Logstash deployment

Why? Planning your deployment strategy ensures optimal coverage, performance, and compatibility with existing infrastructure. Understanding Logstash's architecture and deployment patterns helps you choose the right approach for your environment and log volumes.

Detailed steps:

Identify your deployment pattern:
- Centralized Logstash deployment: Single or clustered Logstash instances receiving logs from multiple sources
  - Best for: Medium to large environments, centralized processing
  - Advantages: Centralized configuration, easier management, resource efficiency
  - Considerations: Network bandwidth, single point of failure, scaling
- Distributed Logstash deployment: Multiple Logstash instances deployed across infrastructure
  - Best for: Large-scale environments, geographically distributed systems
  - Advantages: Reduced network traffic, fault isolation, regional processing
  - Considerations: Configuration management, consistency, monitoring complexity
- Hybrid deployment: Lightweight forwarders (Filebeat, Beats) sending to Logstash aggregators
  - Best for: Optimal resource usage, separation of concerns
  - Advantages: Minimal endpoint footprint, centralized processing, scalability
  - Considerations: Multiple components to manage, network dependencies
- Containerized deployment: Logstash running in Docker or Kubernetes
  - Best for: Cloud-native environments, microservices architectures
  - Advantages: Easy scaling, orchestration integration, resource isolation
  - Considerations: Persistent queue storage, configuration management
Assess existing Logstash deployments:
- Inventory current Logstash installations and versions
- Review existing pipeline configurations
- Identify current output destinations (Elasticsearch, Kafka, etc.)
- Determine if you're migrating completely or implementing multi-output routing
- Document custom filters and transformations
Determine input sources:
- File inputs: Log files via file input plugin
- Beats inputs: Filebeat, Metricbeat, etc.
- Syslog inputs: Network syslog via TCP/UDP
- HTTP inputs: Webhooks and HTTP endpoints
- Kafka inputs: Kafka topics
- TCP/UDP inputs: Network streams
- JDBC inputs: Database queries
- Cloud inputs: AWS S3, Azure, GCP via plugins
Assess resource requirements:
- CPU: 2-8 CPU cores per Logstash instance (varies with pipeline complexity)
- Memory: 2-8 GB heap size (adjust based on throughput and pipeline complexity)
- Disk: 500 MB - 10 GB for persistent queues (based on buffering requirements)
- Network: Sufficient bandwidth for input and output traffic
Plan processing and transformation requirements:
- Identify parsing needs (grok, JSON, CSV, etc.)
- Determine filtering requirements (drop, conditionals)
- Plan field transformations (mutate, date, etc.)
- Design enrichment logic (GeoIP, DNS, lookups)
- Consider performance impact of complex pipelines
Plan for reliability and performance:
- Enable persistent queues for data durability
- Configure dead letter queues for failed events
- Plan for pipeline workers and batch sizes
- Consider multiple pipelines for different log types
Plan configuration management:
- Decide on configuration distribution method (manual, Ansible, Puppet, Chef)
- Plan for configuration versioning and change control
- Determine update and upgrade strategy for Logstash versions
- Consider centralized configuration management

Step 3 - Install Logstash

Why? Installing Logstash on your target systems enables log collection, processing, and forwarding. The installation method varies by platform and deployment pattern, but all methods result in a running Logstash instance ready for configuration.

Detailed steps:

For Debian/Ubuntu Linux systems:

Download and install the public signing key:

wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

Install the apt-transport-https package:

sudo apt-get install apt-transport-https

Add the Elastic repository:

echo "deb https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee -a /etc/apt/sources.list.d/elastic-8.x.list

Update package lists and install Logstash:

sudo apt-get update
sudo apt-get install logstash

Verify the installation:

/usr/share/logstash/bin/logstash --version

For RHEL/CentOS/Fedora systems:

Download and install the public signing key:

sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

Create the Elastic repository file:

cat <<EOF | sudo tee /etc/yum.repos.d/elastic.repo
[elastic-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF

Install Logstash:
```
sudo yum install logstash
```

Verify the installation:

/usr/share/logstash/bin/logstash --version

For Windows systems:
1. Download the Logstash Windows zip file from the Elastic website
2. Extract the contents to C:\logstash
3. Verify the installation:
  C:\logstash\bin\logstash.bat --version
4. Install as a Windows service (optional):
  C:\logstash\bin\logstash-windows-install.bat

For Docker containers:

Pull the official Logstash Docker image:

docker pull docker.elastic.co/logstash/logstash:8.11.0

Run Logstash container with configuration:

docker run -d \
  --name logstash \
  -v /path/to/logstash.conf:/usr/share/logstash/pipeline/logstash.conf \
  -v /path/to/logstash.yml:/usr/share/logstash/config/logstash.yml \
  -p 5044:5044 \
  -p 9600:9600 \
  docker.elastic.co/logstash/logstash:8.11.0

For Kubernetes deployment:

Create a ConfigMap for Logstash configuration:

apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-config
  namespace: logging
data:
  logstash.yml: |
    http.host: "0.0.0.0"
    path.config: /usr/share/logstash/pipeline
  logstash.conf: |
    # Pipeline configuration will be added in Step 4
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: logstash
  namespace: logging
spec:
  replicas: 2
  selector:
    matchLabels:
      app: logstash
  template:
    metadata:
      labels:
        app: logstash
    spec:
      containers:
      - name: logstash
        image: docker.elastic.co/logstash/logstash:8.11.0
        ports:
        - containerPort: 5044
          name: beats
        - containerPort: 9600
          name: http
        volumeMounts:
        - name: config
          mountPath: /usr/share/logstash/config/logstash.yml
          subPath: logstash.yml
        - name: pipeline
          mountPath: /usr/share/logstash/pipeline/logstash.conf
          subPath: logstash.conf
        resources:
          limits:
            memory: 2Gi
            cpu: 1000m
          requests:
            memory: 1Gi
            cpu: 500m
        env:
        - name: LS_JAVA_OPTS
          value: "-Xmx1g -Xms1g"
      volumes:
      - name: config
        configMap:
          name: logstash-config
          items:
          - key: logstash.yml
            path: logstash.yml
      - name: pipeline
        configMap:
          name: logstash-config
          items:
          - key: logstash.conf
            path: logstash.conf
---
apiVersion: v1
kind: Service
metadata:
  name: logstash
  namespace: logging
spec:
  selector:
    app: logstash
  ports:
  - name: beats
    port: 5044
    targetPort: 5044
  - name: http
    port: 9600
    targetPort: 9600

Apply the manifest:

kubectl apply -f logstash-deployment.yaml

Verify deployment:

kubectl get deployment -n logging logstash
kubectl get pods -n logging -l app=logstash

Verify installation across all platforms:
- Check that the Logstash binary is present and executable
- Verify Java is installed (Logstash requires Java 11 or later)
- Confirm configuration directories exist
- Check Logstash version compatibility (8.x recommended)

Step 4 - Configure Logstash to send data to LogScale

Why? Configuration defines what logs Logstash collects, how they're processed, and where they're sent. Proper configuration ensures efficient collection, sophisticated transformation, and reliable delivery to LogScale using Logstash's HTTP output plugin.

Detailed steps:

Locate the Logstash configuration directory:
- Linux: /etc/logstash/conf.d/
- Windows: C:\logstash\config\
- Docker/Kubernetes: Mount configuration via volume or ConfigMap
Understand Logstash pipeline structure:
- input: Defines data sources (where logs come from)
- filter: Defines processing and transformation rules
- output: Defines destinations (where logs go)
- Events flow: input → filter → output

Create a basic Logstash pipeline configuration for LogScale:

# /etc/logstash/conf.d/logscale.conf

input {
  # Example: Beats input (Filebeat, Metricbeat, etc.)
  beats {
    port => 5044
    type => "beats"
  }
  
  # Example: File input
  file {
    path => "/var/log/application/*.log"
    start_position => "beginning"
    sincedb_path => "/var/lib/logstash/sincedb"
    type => "application"
  }
  
  # Example: Syslog input
  syslog {
    port => 5140
    type => "syslog"
  }
}

filter {
  # Add hostname
  mutate {
    add_field => { "hostname" => "%{host}" }
  }
  
  # Parse timestamp
  date {
    match => [ "timestamp", "ISO8601" ]
    target => "@timestamp"
  }
  
  # Add environment tag
  mutate {
    add_field => { "environment" => "production" }
  }
}

output {
  # Output to LogScale
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    
    # Authentication
    headers => {
      "Authorization" => "Bearer YOUR_INGEST_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    
    # Format for LogScale
    format => "json"
    mapping => {
      "tags" => {
        "host" => "%{hostname}"
        "type" => "%{type}"
      }
      "events" => [{
        "timestamp" => "%{@timestamp}"
        "attributes" => "%{message}"
      }]
    }
    
    # Performance settings
    automatic_retries => 3
    retry_non_idempotent => true
    connect_timeout => 10
    socket_timeout => 10
    request_timeout => 60
    
    # Compression
    http_compression => true
  }
  
  # Optional: Also output to stdout for debugging
  # stdout {
  #   codec => rubydebug
  # }
}

Create an optimized configuration using the json_batch codec:

# /etc/logstash/conf.d/logscale-optimized.conf

input {
  beats {
    port => 5044
  }
}

filter {
  # Remove unnecessary fields
  mutate {
    remove_field => [ "@version", "agent", "ecs", "input", "log" ]
  }
  
  # Ensure timestamp is in ISO8601 format
  ruby {
    code => "event.set('timestamp', event.get('@timestamp').to_iso8601)"
  }
}

output {
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    
    headers => {
      "Authorization" => "Bearer YOUR_INGEST_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    
    # Use json_batch codec for better performance
    codec => json_batch {
      batch_separator => ""
    }
    
    # Format events for LogScale structured endpoint
    format => "json_batch"
    content_type => "application/json"
    
    # Batching for performance
    pool_max => 50
    pool_max_per_route => 25
    
    # Retry settings
    automatic_retries => 5
    retry_non_idempotent => true
    
    # Timeouts
    connect_timeout => 10
    socket_timeout => 10
    request_timeout => 60
    
    # Compression
    http_compression => true
  }
}

Configure input-specific pipelines:

For file inputs with grok parsing:

input {
  file {
    path => "/var/log/nginx/access.log"
    start_position => "beginning"
    sincedb_path => "/var/lib/logstash/nginx_sincedb"
    type => "nginx_access"
  }
}

filter {
  if [type] == "nginx_access" {
    grok {
      match => {
        "message" => '%{IPORHOST:remote_addr} - %{DATA:remote_user} \[%{HTTPDATE:time_local}\] "%{WORD:request_method} %{DATA:request_uri} HTTP/%{NUMBER:http_version}" %{NUMBER:status} %{NUMBER:body_bytes_sent} "%{DATA:http_referer}" "%{DATA:http_user_agent}"'
      }
    }
    
    date {
      match => [ "time_local", "dd/MMM/yyyy:HH:mm:ss Z" ]
      target => "@timestamp"
    }
    
    mutate {
      convert => {
        "status" => "integer"
        "body_bytes_sent" => "integer"
      }
    }
  }
}

output {
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer YOUR_INGEST_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    format => "json"
    http_compression => true
  }
}

For JSON logs:

input {
  file {
    path => "/var/log/application/*.json"
    codec => json
    type => "application_json"
  }
}

filter {
  # JSON is already parsed by codec
  # Add any additional fields or transformations
  
  mutate {
    add_field => {
      "environment" => "production"
      "service" => "api"
    }
  }
  
  # Rename fields if needed
  mutate {
    rename => {
      "level" => "log_level"
      "msg" => "message"
    }
  }
}

output {
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer YOUR_INGEST_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    format => "json"
    http_compression => true
  }
}

For syslog with enrichment:

input {
  syslog {
    port => 5140
    type => "syslog"
  }
}

filter {
  if [type] == "syslog" {
    # GeoIP enrichment
    if [host] {
      geoip {
        source => "host"
        target => "geoip"
      }
    }
    
    # DNS lookup
    dns {
      reverse => [ "host" ]
      action => "replace"
    }
    
    # Add custom fields
    mutate {
      add_field => {
        "log_source" => "syslog"
        "facility_label" => "%{syslog_facility_code}"
      }
    }
  }
}

output {
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer YOUR_INGEST_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    format => "json"
    http_compression => true
  }
}

Configure multiple outputs (hybrid approach):

input {
  beats {
    port => 5044
  }
}

filter {
  # Common filtering
  mutate {
    add_field => { "environment" => "production" }
  }
}

output {
  # Send to LogScale
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer YOUR_LOGSCALE_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    format => "json"
    http_compression => true
  }
  
  # Also send to Elasticsearch (for gradual migration)
  elasticsearch {
    hosts => ["elasticsearch.example.com:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
}

Configure conditional routing based on log type:

input {
  beats {
    port => 5044
  }
}

filter {
  # Tag based on log content
  if [message] =~ /ERROR/ {
    mutate {
      add_tag => [ "error" ]
    }
  }
  
  if [message] =~ /security/ {
    mutate {
      add_tag => [ "security" ]
    }
  }
}

output {
  # Send errors to high-priority LogScale repository
  if "error" in [tags] {
    http {
      url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
      http_method => "post"
      headers => {
        "Authorization" => "Bearer YOUR_ERROR_REPO_TOKEN_HERE"
        "Content-Type" => "application/json"
      }
      format => "json"
    }
  }
  
  # Send security logs to security repository
  if "security" in [tags] {
    http {
      url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
      http_method => "post"
      headers => {
        "Authorization" => "Bearer YOUR_SECURITY_REPO_TOKEN_HERE"
        "Content-Type" => "application/json"
      }
      format => "json"
    }
  }
  
  # Send all other logs to general repository
  if "error" not in [tags] and "security" not in [tags] {
    http {
      url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
      http_method => "post"
      headers => {
        "Authorization" => "Bearer YOUR_GENERAL_REPO_TOKEN_HERE"
        "Content-Type" => "application/json"
      }
      format => "json"
    }
  }
}

Configure Logstash settings in logstash.yml:

# /etc/logstash/logstash.yml

# Pipeline settings
pipeline.workers: 4
pipeline.batch.size: 125
pipeline.batch.delay: 50

# Queue settings (persistent queue for reliability)
queue.type: persisted
queue.max_bytes: 1gb
queue.checkpoint.writes: 1024

# Dead letter queue
dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 1gb

# Monitoring
monitoring.enabled: false

# HTTP API
http.host: "0.0.0.0"
http.port: 9600

# Logging
log.level: info
path.logs: /var/log/logstash

Test the configuration:

# Test configuration syntax
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/logscale.conf --config.test_and_exit

# Run with verbose output for debugging
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/logscale.conf --log.level=debug

Step 5 - Configure advanced features and optimization

Why? Advanced configuration optimizes Logstash's performance, reliability, and resource usage based on your specific log volumes, processing complexity, and operational requirements.

Detailed steps:

Configure Persistent Queues for Reliability:
1. Enable persistent queues in logstash.yml:
  # /etc/logstash/logstash.yml queue.type: persisted queue.max_bytes: 2gb queue.page_capacity: 64mb queue.checkpoint.acks: 1024 queue.checkpoint.writes: 1024 queue.checkpoint.interval: 1000 path.queue: /var/lib/logstash/queue
  - Persistent queues protect against data loss during Logstash restarts
  - Events are stored on disk until successfully processed

Configure Dead Letter Queue:

Enable DLQ for failed events:

# /etc/logstash/logstash.yml

dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 1gb
path.dead_letter_queue: /var/lib/logstash/dead_letter_queue

Create a pipeline to process DLQ events:

# /etc/logstash/conf.d/dlq-replay.conf

input {
  dead_letter_queue {
    path => "/var/lib/logstash/dead_letter_queue"
    commit_offsets => true
  }
}

filter {
  # Add DLQ metadata
  mutate {
    add_field => {
      "dlq_reason" => "%{[@metadata][dead_letter_queue][reason]}"
      "dlq_plugin_id" => "%{[@metadata][dead_letter_queue][plugin_id]}"
    }
  }
}

output {
  # Retry sending to LogScale
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer YOUR_INGEST_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    format => "json"
  }
}

Configure Multiple Pipelines:

Define multiple pipelines in pipelines.yml:

# /etc/logstash/pipelines.yml

- pipeline.id: application-logs
  path.config: "/etc/logstash/conf.d/application.conf"
  pipeline.workers: 2
  pipeline.batch.size: 125

- pipeline.id: security-logs
  path.config: "/etc/logstash/conf.d/security.conf"
  pipeline.workers: 4
  pipeline.batch.size: 50
  queue.type: persisted

- pipeline.id: metrics
  path.config: "/etc/logstash/conf.d/metrics.conf"
  pipeline.workers: 1
  pipeline.batch.size: 1000

Multiple pipelines allow different processing for different log types
Each pipeline can have independent performance tuning

Implement Advanced Filtering and Transformation:

Use Ruby filter for complex transformations:

filter {
  ruby {
    code => '
      # Custom transformation logic
      event.set("custom_field", event.get("field1").to_s + "-" + event.get("field2").to_s)
      
      # Conditional logic
      if event.get("status").to_i >= 500
        event.set("severity", "critical")
      elsif event.get("status").to_i >= 400
        event.set("severity", "error")
      else
        event.set("severity", "info")
      end
      
      # Calculate duration
      if event.get("start_time") && event.get("end_time")
        duration = event.get("end_time").to_i - event.get("start_time").to_i
        event.set("duration_ms", duration)
      end
    '
  }
}

Use aggregate filter for multi-line events:

filter {
  # Aggregate multi-line stack traces
  aggregate {
    task_id => "%{thread_id}"
    code => "
      map['message'] ||= ''
      map['message'] += event.get('message') + '\n'
      event.cancel()
    "
    push_map_as_event_on_timeout => true
    timeout_task_id_field => "thread_id"
    timeout => 5
  }
}

Use translate filter for lookups:

filter {
  translate {
    field => "status_code"
    destination => "status_description"
    dictionary_path => "/etc/logstash/dictionaries/http_status.yml"
    fallback => "Unknown status"
  }
}

Optimize Performance:

Tune JVM heap size:

# /etc/logstash/jvm.options

# Set heap size (50-75% of available RAM, max 8GB recommended)
-Xms2g
-Xmx2g

# GC settings for better performance
-XX:+UseG1GC
-XX:MaxGCPauseMillis=200
-XX:InitiatingHeapOccupancyPercent=45
-XX:G1ReservePercent=20

Optimize pipeline settings:

# /etc/logstash/logstash.yml

# Increase workers for CPU-bound workloads
pipeline.workers: 8

# Increase batch size for throughput
pipeline.batch.size: 250

# Reduce batch delay for lower latency
pipeline.batch.delay: 10

# Optimize output workers
pipeline.output.workers: 4

Use pipeline-to-pipeline communication for complex workflows:

# Input pipeline
input {
  beats {
    port => 5044
  }
}

filter {
  # Initial parsing
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
}

output {
  pipeline {
    send_to => ["enrichment_pipeline"]
  }
}

# Enrichment pipeline
input {
  pipeline {
    address => "enrichment_pipeline"
  }
}

filter {
  # Heavy enrichment operations
  geoip {
    source => "clientip"
  }
  
  dns {
    reverse => ["clientip"]
  }
}

output {
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer YOUR_INGEST_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    format => "json"
  }
}

Implement Monitoring and Metrics:

Enable Logstash monitoring API:

# /etc/logstash/logstash.yml

http.host: "0.0.0.0"
http.port: 9600

# Enable monitoring
monitoring.enabled: true

Query monitoring endpoints:

# Node info
curl -X GET "http://localhost:9600/_node?pretty"

# Node stats
curl -X GET "http://localhost:9600/_node/stats?pretty"

# Pipeline stats
curl -X GET "http://localhost:9600/_node/stats/pipelines?pretty"

# Hot threads
curl -X GET "http://localhost:9600/_node/hot_threads?pretty"

Send Logstash metrics to LogScale:

input {
  http_poller {
    urls => {
      logstash_stats => "http://localhost:9600/_node/stats"
    }
    request_timeout => 60
    schedule => { every => "30s" }
    codec => "json"
    metadata_target => "http_poller_metadata"
  }
}

filter {
  # Extract relevant metrics
  ruby {
    code => '
      stats = event.get("pipelines")
      stats.each do |pipeline_id, pipeline_stats|
        # Create separate events for each pipeline metric
        # (implementation details)
      end
    '
  }
}

output {
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer YOUR_METRICS_TOKEN_HERE"
      "Content-Type" => "application/json"
    }
    format => "json"
  }
}

Implement Security Best Practices:

Use Logstash keystore for sensitive data:

# Create keystore
/usr/share/logstash/bin/logstash-keystore create

# Add ingest token to keystore
/usr/share/logstash/bin/logstash-keystore add LOGSCALE_TOKEN

# Use in configuration
output {
  http {
    url => "https://cloud.humio.com/api/v1/ingest/humio-structured"
    http_method => "post"
    headers => {
      "Authorization" => "Bearer ${LOGSCALE_TOKEN}"
      "Content-Type" => "application/json"
    }
    format => "json"
  }
}

Configure TLS for inputs:

input {
  beats {
    port => 5044
    ssl => true
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    ssl_verify_mode => "force_peer"
  }
}

Step 6 - Start Logstash and verify operation

Why? Starting Logstash activates log collection, processing, and forwarding. Verification ensures Logstash is functioning correctly and data is flowing to LogScale as expected.

Detailed steps:

Start the Logstash service:

For Linux (systemd):

sudo systemctl start logstash
sudo systemctl enable logstash  # Enable auto-start on boot

For Linux (init.d):

sudo service logstash start
sudo chkconfig logstash on  # Enable auto-start on boot

For Windows:

net start logstash
# Or use Services management console (services.msc)

For Docker:
```
docker start logstash
```

For Kubernetes:

# Deployment starts automatically
kubectl rollout status deployment/logstash -n logging

Verify Logstash is running:

For Linux:

sudo systemctl status logstash
# Check for "active (running)" status

For Windows:

sc query logstash
# Check for "RUNNING" state

For Docker:

docker ps | grep logstash
docker logs logstash

For Kubernetes:

kubectl get pods -n logging -l app=logstash
kubectl logs -n logging -l app=logstash

Check Logstash logs for startup messages:

Log location: /var/log/logstash/logstash-plain.log
Look for successful pipeline startup
Verify plugins are loaded correctly
Confirm input sources are initialized
Check for any error or warning messages

Example log entries to look for:

[INFO ][logstash.runner          ] Starting Logstash
[INFO ][logstash.agent           ] Successfully started Logstash API endpoint
[INFO ][logstash.javapipeline    ] Starting pipeline {:pipeline_id=>"main"}
[INFO ][logstash.inputs.beats    ] Beats inputs: Starting input listener {:address=>"0.0.0.0:5044"}
[INFO ][logstash.javapipeline    ] Pipeline started {"pipeline.id"=>"main"}

Verify input sources are active:
- Check listening ports:
  sudo netstat -tlnp | grep java # or sudo ss -tlnp | grep java
- Verify file inputs are monitoring files
- Check API endpoint is accessible:
  curl http://localhost:9600/?pretty

Generate test log entries:

Write test entries to monitored log files:

echo "Test log entry from Logstash $(date)" >> /var/log/application/test.log

Send test message via Beats:

# From a Filebeat instance
filebeat test output

Send test syslog message:

logger -n localhost -P 5140 -T "Test syslog message"

Verify data in LogScale:
- Navigate to your repository in LogScale
- Run a query to find recently ingested events from Logstash:
  @source=logstash OR host=*
- Verify events have correct timestamps
- Confirm fields are extracted correctly
- Check that custom fields and tags are present
- Verify host metadata is included
Check Logstash metrics via API:
- Get pipeline statistics:
  curl -X GET "http://localhost:9600/_node/stats/pipelines?pretty"
- Review key metrics:
  - events.in: Events received by pipeline
  - events.out: Events sent from pipeline
  - events.filtered: Events processed by filters
  - events.duration_in_millis: Processing time
  - queue.events: Events in queue
Verify continuous operation:
- Monitor for several minutes to ensure stable operation
- Check that events continue to flow to LogScale
- Verify no error accumulation in Logstash logs
- Confirm resource usage (CPU, memory) is within acceptable limits
Test error handling and recovery:
- Temporarily block network access to LogScale endpoint
- Verify events are queued in persistent queue
- Monitor queue size growth
- Restore network access and confirm queued events are sent
- Verify no data loss occurred during the outage
Test Logstash restart behavior:
- Restart the Logstash service
- Verify it resumes from the last processed position
- Confirm persistent queue is preserved
- Check that no events are lost or duplicated

Step 7 - Monitoring and maintenance

Why? Ongoing monitoring ensures the reliability, performance, and efficiency of your Logstash infrastructure, enabling proactive issue detection and continuous optimization of log processing pipelines.

What you should do:

Set up alerts for Logstash health:
- Logstash process/service down or not running
- Pipeline failures or stalls
- Persistent queue size exceeding thresholds
- Dead letter queue accumulation
- Output plugin errors or connection failures
- High CPU or memory usage
- JVM heap usage approaching limits
- Logstash version outdated or unsupported
Monitor Logstash performance metrics:
- Events in vs. events out (should be equal after filtering)
- Event processing rate (events per second)
- Pipeline throughput and latency
- Filter execution time per plugin
- Queue utilization and depth
- Worker thread utilization
- JVM heap usage and garbage collection
- CPU and memory consumption
Monitor data quality and completeness:
- Compare events processed vs. events sent to LogScale
- Check for gaps in log timestamps
- Verify parsing success rates
- Monitor for grok parse failures
- Validate field extraction accuracy
Create operational dashboards:
- Logstash fleet health overview (all instances)
- Per-instance Logstash status and metrics
- Pipeline throughput trends by type
- Queue utilization and flush patterns
- Error and warning summaries with drill-down
- Resource utilization across the Logstash fleet
- Plugin performance breakdown
Regularly review and optimize:
- Pipeline configurations based on actual log patterns
- Filter efficiency and complexity
- Grok patterns for performance
- Worker count and batch sizes
- Queue sizes based on throughput and reliability needs
- JVM heap size based on actual usage
- Plugin versions and updates
Maintain Logstash fleet consistency:
- Use configuration management tools (Ansible, Puppet, Chef)
- Version control all pipeline configurations
- Implement configuration validation before deployment
- Document configuration standards and patterns
- Maintain inventory of all Logstash deployments
Implement update and upgrade procedures:
- Monitor Elastic release notes for new Logstash versions
- Test new versions in non-production environments
- Plan phased rollouts to minimize risk
- Verify plugin compatibility with new versions
- Test configuration compatibility across versions
- Maintain rollback procedures
Implement security best practices:
- Rotate ingest tokens regularly (every 90 days recommended)
- Use Logstash keystore for sensitive credentials
- Run Logstash with minimal required permissions
- Enable TLS for all network inputs
- Secure the Logstash API endpoint
- Audit pipeline configurations for security issues
- Monitor for unauthorized access attempts
Handle log source changes:
- Update grok patterns when log formats change
- Adjust filters when field names change
- Test pipeline changes before production deployment
- Document all configuration changes
Manage persistent queues and DLQ:
- Monitor queue directory disk usage
- Regularly review and process DLQ events
- Clean up old queue data periodically
- Back up queue data before major changes
Plan for scaling and growth:
- Monitor log volume trends
- Plan Logstash capacity for growth
- Consider horizontal scaling with multiple instances
- Implement load balancing for inputs
- Test disaster recovery procedures
- Document architecture and dependencies
Troubleshoot common issues:
- No data appearing in LogScale:
  - Verify Logstash is running and pipelines are active
  - Check ingest token is valid
  - Verify network connectivity to LogScale
  - Check Logstash logs for output errors
  - Verify input sources are receiving data
- High CPU usage:
  - Review filter complexity (especially grok patterns)
  - Optimize Ruby filters
  - Reduce worker count if over-provisioned
  - Consider splitting pipelines
- High memory usage:
  - Reduce JVM heap size if over-allocated
  - Decrease batch size
  - Reduce queue size
  - Check for memory leaks in custom filters
- Pipeline stalls:
  - Check for blocking filters
  - Verify output connectivity
  - Review hot threads: curl http://localhost:9600/_node/hot_threads
  - Check for deadlocks in custom code
- Parsing failures:
  - Test grok patterns with sample data
  - Check for log format changes
  - Review _grokparsefailure tags in events
  - Use grok debugger for pattern development
Maintain operational documentation:
- Document all Logstash deployments and purposes
- Maintain runbooks for common issues
- Document pipeline configurations and logic
- Keep inventory of monitored log sources
- Document custom filters and plugins
- Maintain change logs
Conduct regular reviews:
- Quarterly reviews of all Logstash deployments
- Monthly performance reviews and optimization
- Weekly operational reviews of errors
- Annual disaster recovery testing
Leverage community resources:
- Review Elastic documentation for updates
- Participate in Elastic community forums
- Explore community plugins and patterns
- Share lessons learned
Consider migration strategies:
- If migrating from Elasticsearch to LogScale, plan phased approach
- Test multi-output configurations
- Validate data completeness during migration
- Document field mapping differences
- Plan for eventual removal of legacy outputs

Versions of this Page

Getting Data In

How is Data Impacted?

How to Get Data in

Manage Data Ingest

Package Marketplace

Package Reference

Dashboard Reference

Package Management

Other Integrations

Log Formats

Set up Logstash as an ingest method

Enter search term