Netscaler日志ELK摄入管道性能问题排查与优化需求

问题背景 #

将NetScaler设备的系统日志（syslog）发送至ELK栈，通过自定义摄入管道完成日志解析。虽然管道能实现预期解析功能，但部署后出现Ingest节点负载过高、Logstash数据排队的问题，进而影响整个ELK栈性能，目前已暂停NetScaler日志的加载。需分析现有摄入管道配置，提供性能优化方案。

原配置信息 #

1. 原Ingest管道模拟配置 #

POST _ingest/pipeline/_simulate
{
"pipeline": {
    "description" : "Netscaler logs",
    "processors" :[
  {
    "json": {
      "field": "message",
      "add_to_root": false,
      "target_field": "qrapp"
    }
  },
   {
    "grok": {
      "field": "qrapp.message",
      "patterns": [
        
        "%{DATA:qrapp.eventid} %{DATA:qrapp.event.timestamp} %{DATA:qrapp.event.timezone} %{DATA:qrapp.hostname} : %{CISCO_REASON:qrapp.msg1} : Context %{DATA:qrapp.context_user}@%{IP:qrapp.context_ip} - %{GREEDYDATA:qrapp.msg2}",
        "%{DATA:qrapp.eventid} %{DATA:qrapp.event.timestamp} %{DATA:qrapp.event.timezone} %{DATA:qrapp.hostname} : %{CISCO_REASON:qrapp.msg1} :%{SPACE}%{GREEDYDATA:qrapp.msg2}",
        "%{DATA:qrapp.eventid} %{DATA:qrapp.event.timestamp} %{DATA:qrapp.event.timezone} %{DATA:qrapp.hostname} : %{GREEDYDATA:qrapp.msg}",
        "%{GREEDYDATA:qrapp.msg}"
      ],
      "ignore_missing": true
    }
  },
   {
    "kv": {
      "field": "qrapp.msg2",
      "field_split": " - ",
      "value_split": " ",
      "target_field": "qrapp",
      "ignore_missing": true,
      "ignore_failure": true
    }
  },
  {
    "gsub": {
      "field": "qrapp.eventid",
      "pattern": "[\\<\\>]",
      "replacement": "",
      "ignore_missing": true,
      "ignore_failure": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Total_bytes_recv",
      "target_field": "qrapp.total_bytes_recv",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Total_bytes_send",
      "target_field": "qrapp.total_bytes_send",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Destination",
      "target_field": "qrapp.destination",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Nat_ip",
      "target_field": "qrapp.nat_ip",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.User",
      "target_field": "qrapp.user",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Vserver",
      "target_field": "qrapp.vserver",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Access",
      "target_field": "qrapp.access",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Source",
      "target_field": "qrapp.source",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Client_ip",
      "target_field": "qrapp.client_ip",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Compression_ratio_send",
      "target_field": "qrapp.compression_ratio_send",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Total_compressedbytes_send",
      "target_field": "qrapp.total_compressedbytes_send",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.End_time",
      "target_field": "qrapp.end_time",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Start_time",
      "target_field": "qrapp.start_time",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Duration",
      "target_field": "qrapp.duration",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Total_compressedbytes_recv",
      "target_field": "qrapp.total_compressedbytes_recv",
      "ignore_missing": true
    }
  },
  {
    "rename": {
      "field": "qrapp.Compression_ratio_recv",
      "target_field": "qrapp.compression_ratio_recv",
      "ignore_missing": true
    }
  },
  {
    "date": {
      "formats": [
        "yyyy-MM-dd_HH:mm:ss.SSS",
        "dd-MMM-yy_HH.mm.ss.SSS",
        "yyyy-MM-d_HH:mm:ss.SSS",
        "yyyy-MM-dd HH:mm:ss,SSS",
        "yyyy-MM-d HH:mm:ss,SSS"
      ],
      "ignore_failure": true,
      "field": "qrapp.timestamp",
      "target_field": "@timestamp",
      "timezone": "Asia/Qatar"
    }
  },
  {
    "remove": {
      "field": ["qrapp.message", "message", "qrapp.msg2", "qrapp.context_ip", "qrapp.context_user"],
      "ignore_missing": true
    }
  }
]
},
    "docs": [
     {
       "_source": {
         "message": "\"{\"@version\":\"1\",\"tags\":[\"_grokparsefailure_sysloginput\"],\"message\":\"<134> 11/07/2022:08:02:00 GMT ************* 0-PPE-0 : default SSLVPN ICAEND_CONNSTAT 1314466465 0 :  Source *********:*** - Destination *********:*** - customername  - username:domainname *****@******: - startTime \\\"11/07/2022:07:57:46 GMT\\\" - endTime \\\"11/07/2022:08:02:00 GMT\\\" - Duration 00:04:14  - Total_bytes_send 73134 - Total_bytes_recv 44465 - Total_compressedbytes_send 0 - Total_compressedbytes_recv 0 - Compression_ratio_send 0.00% - Compression_ratio_recv 0.00% - connectionId 47345284 - Total_bytes_wire_send 4710555505001451848 - Total_bytes_wire_recv 5931570851812355409\n\",\"severity_label\":\"Emergency\",\"@timestamp\":\"2022-11-07T08:00:07.384Z\",\"host\":\"********\",\"facility_label\":\"kernel\",\"severity\":0,\"priority\":0,\"facility\":0}\""
         }
     }
   ]
}

2. 原Logstash输入输出配置 #

input {
  # Section for Netscaler
  syslog {
    port => ****
    id => "check-syslog"
   }
}

# output section
output {
  # Section for Netscaler
  if [host] == '********' {
    file {
     path => "*************.log"
    }
  }
}

性能瓶颈分析 #

1. Grok模式低效：
   • 大量使用DATA匹配任意字符串，导致Grok引擎频繁回溯；GREEDYDATA会扫描整个剩余字符串，匹配失败时开销极大。
   • 模式顺序不合理，兜底的%{GREEDYDATA:qrapp.msg}放在最后，但前三个模式的匹配精度不足，导致大量日志需要多次尝试匹配。
2. 重复Rename处理器：每个字段单独使用一个rename处理器，会多次遍历整个文档，增加CPU占用。
3. Date处理器无效：配置中date处理器引用的qrapp.timestamp字段不存在（实际Grok提取的是qrapp.event.timestamp），导致该处理器每次执行都失败，浪费资源。
4. KV处理器无过滤：kv处理器未指定需要提取的字段，会解析所有-分割的键值对，包括不需要的字段，增加处理量。
5. 冗余JSON解析：原日志的message字段是嵌套JSON，Logstash已经做过一次syslog解析，Ingest管道再次解析JSON，存在重复处理。

优化方案 #

一、Ingest管道优化 #

1. 优化Grok模式

• 优先匹配最精确的日志格式，减少匹配尝试次数；用更精确的模式替换DATA，比如用SYSLOGPRI匹配日志优先级，自定义时间模式匹配NetScaler的时间格式。
• 移除不必要的GREEDYDATA，仅在最后作为兜底。

2. 合并Rename操作

使用script处理器一次性完成所有字段的大小写转换，替代多个独立的rename处理器，减少文档遍历次数。

3. 修复Date处理器

修正字段引用为qrapp.event.timestamp，并只保留NetScaler实际输出的时间格式，减少格式尝试次数。

4. 优化KV处理器

指定include_keys仅提取需要的字段，避免处理无关键值对；调整分割符避免误分割。

5. 提前清理无用字段

在处理流程早期移除不需要的字段，减少后续处理器的数据处理量。

优化后的Ingest管道示例

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "description": "Optimized Netscaler logs pipeline",
    "processors": [
      {
        "json": {
          "field": "message",
          "add_to_root": false,
          "target_field": "qrapp",
          "ignore_failure": true
        }
      },
      {
        "grok": {
          "field": "qrapp.message",
          "patterns": [
            "<%{SYSLOGPRI:qrapp.syslog_pri}> %{DATA:qrapp.event.timestamp} %{DATA:qrapp.event.timezone} %{DATA:qrapp.hostname} : %{DATA:qrapp.msg1} : Context %{DATA:qrapp.context_user}@%{IP:qrapp.context_ip} - %{DATA:qrapp.msg2}",
            "<%{SYSLOGPRI:qrapp.syslog_pri}> %{DATA:qrapp.event.timestamp} %{DATA:qrapp.event.timezone} %{DATA:qrapp.hostname} : %{DATA:qrapp.msg1} :%{SPACE}%{DATA:qrapp.msg2}",
            "<%{SYSLOGPRI:qrapp.syslog_pri}> %{DATA:qrapp.event.timestamp} %{DATA:qrapp.event.timezone} %{DATA:qrapp.hostname} : %{GREEDYDATA:qrapp.msg}",
            "%{GREEDYDATA:qrapp.msg}"
          ],
          "ignore_missing": true,
          "match_cache_size": 1000
        }
      },
      {
        "gsub": {
          "field": "qrapp.eventid",
          "pattern": "[<>]",
          "replacement": "",
          "ignore_missing": true
        }
      },
      {
        "kv": {
          "field": "qrapp.msg2",
          "field_split": " - ",
          "value_split": " ",
          "target_field": "qrapp",
          "ignore_missing": true,
          "include_keys": [
            "Source", "Destination", "startTime", "endTime", "Duration",
            "Total_bytes_send", "Total_bytes_recv", "Total_compressedbytes_send",
            "Total_compressedbytes_recv", "Compression_ratio_send", "Compression_ratio_recv"
          ]
        }
      },
      {
        "script": {
          "source": """
            def renameFields = [
              "Total_bytes_recv": "total_bytes_recv",
              "Total_bytes_send": "total_bytes_send",
              "Destination": "destination",
              "Nat_ip": "nat_ip",
              "User": "user",
              "Vserver": "vserver",
              "Access": "access",
              "Source": "source",
              "Client_ip": "client_ip",
              "Compression_ratio_send": "compression_ratio_send",
              "Total_compressedbytes_send": "total_compressedbytes_send",
              "End_time": "end_time",
              "Start_time": "start_time",
              "Duration": "duration",
              "Total_compressedbytes_recv": "total_compressedbytes_recv",
              "Compression_ratio_recv": "compression_ratio_recv"
            ];
            for (Map.Entry entry : renameFields.entrySet()) {
              def oldField = "qrapp." + entry.getKey();
              def newField = "qrapp." + entry.getValue();
              if (ctx.containsKey(oldField)) {
                ctx[newField] = ctx.remove(oldField);
              }
            }
          """,
          "ignore_failure": true
        }
      },
      {
        "date": {
          "formats": ["dd/MM/yyyy:HH:mm:ss z"],
          "ignore_failure": true,
          "field": "qrapp.event.timestamp",
          "target_field": "@timestamp",
          "timezone": "Asia/Qatar"
        }
      },
      {
        "remove": {
          "field": ["qrapp.message", "message", "qrapp.msg2", "qrapp.context_ip", "qrapp.context_user"],
          "ignore_missing": true
        }
      }
    ]
  },
  "docs": [
    {
      "_source": {
        "message": "\"{\"@version\":\"1\",\"tags\":[\"_grokparsefailure_sysloginput\"],\"message\":\"<134> 11/07/2022:08:02:00 GMT ************* 0-PPE-0 : default SSLVPN ICAEND_CONNSTAT 1314466465 0 :  Source *********:*** - Destination *********:*** - customername  - username:domainname *****@******: - startTime \\\"11/07/2022:07:57:46 GMT\\\" - endTime \\\"11/07/2022:08:02:00 GMT\\\" - Duration 00:04:14  - Total_bytes_send 73134 - Total_bytes_recv 44465 - Total_compressedbytes_send 0 - Total_compressedbytes_recv 0 - Compression_ratio_send 0.00% - Compression_ratio_recv 0.00% - connectionId 47345284 - Total_bytes_wire_send 4710555505001451848 - Total_bytes_wire_recv 5931570851812355409\n\",\"severity_label\":\"Emergency\",\"@timestamp\":\"2022-11-07T08:00:07.384Z\",\"host\":\"********\",\"facility_label\":\"kernel\",\"severity\":0,\"priority\":0,\"facility\":0}\""
      }
    }
  ]
}

二、Logstash配置优化 #

1. 启用ECS兼容性：在syslog输入中启用ecs_compatibility => v8，让Logstash提前解析syslog字段，减少Ingest管道的处理压力。
2. 直接输出到Elasticsearch：避免输出到文件再导入，直接将NetScaler日志发送到Elasticsearch，减少中间环节。
3. 调整Pipeline参数：根据服务器CPU核心数设置pipeline.workers（建议等于CPU核心数），提升Logstash的处理能力。

优化后的Logstash配置示例：