Stack monitoring - logstash 8.0 metricbeat standalone compatibility

x-pack/plugins/monitoring/common/types/es.ts

@@ -586,6 +586,17 @@ export interface ElasticsearchMetricbeatSource {
   };
   logstash?: {
     node?: {
+      state?: {
+        pipeline?: {
+          id: string;
+          name: string;
+          representation?: {
+            graph?: {
+              vertices: ElasticsearchSourceLogstashPipelineVertex[];
+            };
+          };
+        };
+      };
       stats?: {
         timestamp?: string;
         logstash?: {

x-pack/plugins/monitoring/public/application/pages/logstash/pipeline.tsx

@@ -142,11 +142,12 @@ export const LogStashPipelinePage: React.FC<ComponentProps> = ({ clusters }) =>
     }
   }, [detailVertexId, getPageData]);
 
-  const onChangePipelineHash = useCallback(() => {
-    window.location.hash = getSafeForExternalLink(
-      `#/logstash/pipelines/${pipelineId}/${pipelineHash}`
-    );
-  }, [pipelineId, pipelineHash]);
+  const onChangePipelineHash = useCallback(
+    (hash) => {
+      window.location.hash = getSafeForExternalLink(`#/logstash/pipelines/${pipelineId}/${hash}`);
+    },
+    [pipelineId]
+  );
 
   useEffect(() => {
     if (cluster) {

x-pack/plugins/monitoring/public/application/pages/logstash/pipeline_versions_dropdown.tsx

@@ -12,7 +12,7 @@ import React from 'react';
 interface Props {
   pipelineVersions: any[];
   pipelineHash?: string;
-  onChangePipelineHash: () => void;
+  onChangePipelineHash: (hash: string) => void;
 }
 
 export const PipelineVersions = (props: Props) => {
@@ -23,23 +23,25 @@ export const PipelineVersions = (props: Props) => {
       <EuiFlexItem grow={false}>
         <EuiSelect
           value={pipelineHash}
-          options={pipelineVersions.map((option) => {
-            return {
-              text: i18n.translate(
-                'xpack.monitoring.logstashNavigation.pipelineVersionDescription',
-                {
-                  defaultMessage:
-                    'Version active {relativeLastSeen} and first seen {relativeFirstSeen}',
-                  values: {
-                    relativeLastSeen: option.relativeLastSeen,
-                    relativeFirstSeen: option.relativeFirstSeen,
-                  },
-                }
-              ),
-              value: option.hash,
-            };
-          })}
-          onChange={onChangePipelineHash}
+          options={pipelineVersions.map(
+            (option: { hash: string; relativeLastSeen: number; relativeFirstSeen: number }) => {
+              return {
+                text: i18n.translate(
+                  'xpack.monitoring.logstashNavigation.pipelineVersionDescription',
+                  {
+                    defaultMessage:
+                      'Version active {relativeLastSeen} and first seen {relativeFirstSeen}',
+                    values: {
+                      relativeLastSeen: option.relativeLastSeen,
+                      relativeFirstSeen: option.relativeFirstSeen,
+                    },
+                  }
+                ),
+                value: option.hash,
+              };
+            }
+          )}
+          onChange={({ target }) => onChangePipelineHash(target.value)}
         />
       </EuiFlexItem>
     </EuiFlexGroup>

x-pack/plugins/monitoring/server/lib/logstash/get_logstash_for_clusters.ts

@@ -56,7 +56,7 @@ export function getLogstashForClusters(
         ignore_unavailable: true,
         body: {
           query: createQuery({
-            types: ['stats', 'logstash_stats'],
+            types: ['node_stats', 'logstash_stats'],
             start,
             end,
             clusterUuid,

x-pack/plugins/monitoring/server/lib/logstash/get_nodes.ts

@@ -85,7 +85,7 @@ export async function getNodes(
         end,
         clusterUuid,
         metric: LogstashMetric.getMetricFields(),
-        types: ['stats', 'logstash_stats'],
+        types: ['node_stats', 'logstash_stats'],
       }),
       collapse: {
         field: 'logstash_stats.logstash.uuid',

x-pack/plugins/monitoring/server/lib/logstash/get_paginated_pipelines.ts

@@ -154,7 +154,7 @@ async function getPaginatedThroughputData(
                       },
                       {
                         term: {
-                          'metricset.name': 'stats',
+                          'metricset.name': 'node_stats',
                         },
                       },
                     ],
@@ -208,7 +208,10 @@ async function getPaginatedNodesData(
     [
       {
         bool: {
-          should: [{ term: { type: 'logstash_stats' } }, { term: { 'metricset.name': 'stats' } }],
+          should: [
+            { term: { type: 'logstash_stats' } },
+            { term: { 'metricset.name': 'node_stats' } },
+          ],
         },
       },
     ],
@@ -296,7 +299,7 @@ async function getThroughputPipelines(
                     },
                     {
                       term: {
-                        'metricset.name': 'stats',
+                        'metricset.name': 'node_stats',
                       },
                     },
                   ],
@@ -330,7 +333,10 @@ async function getNodePipelines(
     [
       {
         bool: {
-          should: [{ term: { type: 'logstash_stats' } }, { term: { 'metricset.name': 'stats' } }],
+          should: [
+            { term: { type: 'logstash_stats' } },
+            { term: { 'metricset.name': 'node_stats' } },
+          ],
         },
       },
     ],

x-pack/plugins/monitoring/server/lib/logstash/get_pipeline.ts

@@ -76,7 +76,7 @@ export function _enrichStateWithStatsAggregation(
   statsAggregation: any,
   timeseriesIntervalInSeconds: number
 ) {
-  const logstashState = stateDocument.logstash_state;
+  const logstashState = stateDocument.logstash_state || stateDocument.logstash?.node?.state;
   const vertices = logstashState?.pipeline?.representation?.graph?.vertices ?? [];
 
   const verticesById: any = {};
@@ -90,7 +90,9 @@ export function _enrichStateWithStatsAggregation(
   const totalProcessorsDurationInMillis = totalDurationStats.max - totalDurationStats.min;
 
   const verticesWithStatsBuckets =
-    statsAggregation.aggregations?.pipelines.scoped.vertices.vertex_id.buckets ?? [];
+    statsAggregation.aggregations?.pipelines.scoped.vertices?.vertex_id.buckets ??
+    statsAggregation.aggregations?.pipelines_mb.scoped.vertices?.vertex_id.buckets ??
+    [];
   verticesWithStatsBuckets.forEach((vertexStatsBucket: any) => {
     // Each vertexStats bucket contains a list of stats for a single vertex within a single timeseries interval
     const vertexId = vertexStatsBucket.key;
@@ -107,7 +109,7 @@ export function _enrichStateWithStatsAggregation(
     }
   });
 
-  return stateDocument.logstash_state?.pipeline;
+  return logstashState?.pipeline;
 }
 
 export async function getPipeline(
@@ -121,7 +123,7 @@ export async function getPipeline(
   checkParam(lsIndexPattern, 'lsIndexPattern in getPipeline');
 
   // Determine metrics' timeseries interval based on version's timespan
-  const minIntervalSeconds = config.get('monitoring.ui.min_interval_seconds');
+  const minIntervalSeconds = Math.max(Number(config.get('monitoring.ui.min_interval_seconds')), 30);
   const timeseriesInterval = calculateTimeseriesInterval(
     Number(version.firstSeen),
     Number(version.lastSeen),

x-pack/plugins/monitoring/server/lib/logstash/get_pipeline_state_document.ts

@@ -35,7 +35,7 @@ export async function getPipelineStateDocument({
     // This is important because a user may pick a very narrow time picker window. If we were to use a start/end value
     // that could result in us being unable to render the graph
     // Use the logstash_stats documents to determine whether the instance is up/down
-    type: 'logstash_state',
+    types: ['logstash_state', 'node'],
     metric: LogstashMetric.getMetricFields(),
     clusterUuid,
     filters,

x-pack/plugins/monitoring/server/lib/logstash/get_pipeline_stats_aggregation.ts

@@ -52,16 +52,16 @@ function scalarCounterAggregation(
   return aggs;
 }
 
-function nestedVertices(maxBucketSize: string) {
-  const fieldPath = 'logstash_stats.pipelines.vertices';
-  const ephemeralIdField = 'logstash_stats.pipelines.vertices.pipeline_ephemeral_id';
+function nestedVertices(statsPath: string, maxBucketSize: string) {
+  const fieldPath = `${statsPath}.pipelines.vertices`;
+  const ephemeralIdField = `${statsPath}.pipelines.vertices.pipeline_ephemeral_id`;
 
   return {
-    nested: { path: 'logstash_stats.pipelines.vertices' },
+    nested: { path: `${statsPath}.pipelines.vertices` },
     aggs: {
       vertex_id: {
         terms: {
-          field: 'logstash_stats.pipelines.vertices.id',
+          field: `${statsPath}.pipelines.vertices.id`,
           size: maxBucketSize,
         },
         aggs: {
@@ -79,24 +79,33 @@ function nestedVertices(maxBucketSize: string) {
   };
 }
 
-function createScopedAgg(pipelineId: string, pipelineHash: string, agg: { [key: string]: any }) {
-  return {
-    pipelines: {
-      nested: { path: 'logstash_stats.pipelines' },
+function createScopedAgg(pipelineId: string, pipelineHash: string, maxBucketSize: string) {
+  return (statsPath: string) => {
+    const verticesAgg = {
+      vertices: nestedVertices(statsPath, maxBucketSize),
+      total_processor_duration_stats: {
+        stats: {
+          field: `${statsPath}.pipelines.events.duration_in_millis`,
+        },
+      },
+    };
+
+    return {
+      nested: { path: `${statsPath}.pipelines` },
       aggs: {
         scoped: {
           filter: {
             bool: {
               filter: [
-                { term: { 'logstash_stats.pipelines.id': pipelineId } },
-                { term: { 'logstash_stats.pipelines.hash': pipelineHash } },
+                { term: { [`${statsPath}.pipelines.id`]: pipelineId } },
+                { term: { [`${statsPath}.pipelines.hash`]: pipelineHash } },
               ],
             },
           },
-          aggs: agg,
+          aggs: verticesAgg,
         },
       },
-    },
+    };
   };
 }
 
@@ -109,6 +118,7 @@ function fetchPipelineLatestStats(
   callWithRequest: any,
   req: LegacyRequest
 ) {
+  const pipelineAggregation = createScopedAgg(pipelineId, version.hash, maxBucketSize);
   const params = {
     index: logstashIndexPattern,
     size: 0,
@@ -119,17 +129,18 @@ function fetchPipelineLatestStats(
       'aggregations.pipelines.scoped.vertices.vertex_id.buckets.events_out_total',
       'aggregations.pipelines.scoped.vertices.vertex_id.buckets.duration_in_millis_total',
       'aggregations.pipelines.scoped.total_processor_duration_stats',
+      'aggregations.pipelines_mb.scoped.vertices.vertex_id.buckets.key',
+      'aggregations.pipelines_mb.scoped.vertices.vertex_id.buckets.events_in_total',
+      'aggregations.pipelines_mb.scoped.vertices.vertex_id.buckets.events_out_total',
+      'aggregations.pipelines_mb.scoped.vertices.vertex_id.buckets.duration_in_millis_total',
+      'aggregations.pipelines_mb.scoped.total_processor_duration_stats',
     ],
     body: {
       query,
-      aggs: createScopedAgg(pipelineId, version.hash, {
-        vertices: nestedVertices(maxBucketSize),
-        total_processor_duration_stats: {
-          stats: {
-            field: 'logstash_stats.pipelines.events.duration_in_millis',
-          },
-        },
-      }),
+      aggs: {
+        pipelines: pipelineAggregation('logstash_stats'),
+        pipelines_mb: pipelineAggregation('logstash.node.stats'),
+      },
     },
   };
 
@@ -154,16 +165,31 @@ export function getPipelineStatsAggregation({
   const { callWithRequest } = req.server.plugins.elasticsearch.getCluster('monitoring');
   const filters = [
     {
-      nested: {
-        path: 'logstash_stats.pipelines',
-        query: {
-          bool: {
-            must: [
-              { term: { 'logstash_stats.pipelines.hash': version.hash } },
-              { term: { 'logstash_stats.pipelines.id': pipelineId } },
-            ],
+      bool: {
+        should: [
+          {
+            nested: {
+              path: 'logstash_stats.pipelines',
+              ignore_unmapped: true,
+              query: {
+                bool: {
+                  filter: [{ term: { 'logstash_stats.pipelines.id': pipelineId } }],
+                },
+              },
+            },
           },
-        },
+          {
+            nested: {
+              path: 'logstash.node.stats.pipelines',
+              ignore_unmapped: true,
+              query: {
+                bool: {
+                  filter: [{ term: { 'logstash.node.stats.pipelines.id': pipelineId } }],
+                },
+              },
+            },
+          },
+        ],
       },
     },
   ];
@@ -172,7 +198,7 @@ export function getPipelineStatsAggregation({
   const end = version.lastSeen;
 
   const query = createQuery({
-    types: ['stats', 'logstash_stats'],
+    types: ['node_stats', 'logstash_stats'],
     start,
     end,
     metric: LogstashMetric.getMetricFields(),