[Fleet] Improve validation for dynamic Kafka topics

x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/settings/components/edit_output_flyout/output_form_kafka_topics.tsx

@@ -15,6 +15,7 @@ import {
   EuiTitle,
   EuiRadioGroup,
   EuiComboBox,
+  EuiLink,
 } from '@elastic/eui';
 import { FormattedMessage } from '@kbn/i18n-react';
 import { i18n } from '@kbn/i18n';
@@ -33,7 +34,7 @@ export const OutputFormKafkaTopics: React.FunctionComponent<{ inputs: OutputForm
   const dynamicOptions: Array<EuiComboBoxOptionOption<string>> = useMemo(() => {
     const options = KAFKA_DYNAMIC_FIELDS.map((option) => ({
       label: option,
-      value: option,
+      value: `%{[${option}]}`,
     }));
     return options;
   }, []);
@@ -73,7 +74,21 @@ export const OutputFormKafkaTopics: React.FunctionComponent<{ inputs: OutputForm
             label={
               <FormattedMessage
                 id="xpack.fleet.settings.editOutputFlyout.kafkaDynamicTopicLabel"
-                defaultMessage="Topic from field"
+                defaultMessage="Topic from field(s). For more info, see our {guideLink}"
+                values={{
+                  guideLink: (
+                    <EuiLink
+                      href="https://www.elastic.co/guide/en/beats/filebeat/current/kafka-output.html#topic-option-kafka"
+                      target="_blank"
+                      external
+                    >
+                      <FormattedMessage
+                        id="xpack.fleet.settings.kafkaGuideLink"
+                        defaultMessage="docs."
+                      />
+                    </EuiLink>
+                  ),
+                }}
               />
             }
             {...inputs.kafkaDynamicTopicInput.formRowProps}
@@ -83,7 +98,7 @@ export const OutputFormKafkaTopics: React.FunctionComponent<{ inputs: OutputForm
               fullWidth
               isClearable={true}
               options={dynamicOptions}
-              customOptionText="Use custom field (not recommended)"
+              customOptionText="Use custom field"
               singleSelection={{ asPlainText: true }}
               {...inputs.kafkaDynamicTopicInput.props}
             />

x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/settings/components/edit_output_flyout/output_form_validators.test.tsx

@@ -14,6 +14,7 @@ import {
   validateKafkaHosts,
   validateKibanaURL,
   validateKibanaAPIKey,
+  validateDynamicKafkaTopics,
 } from './output_form_validators';
 
 describe('Output form validation', () => {
@@ -336,4 +337,26 @@ describe('Output form validation', () => {
       ]);
     });
   });
+
+  describe('validateDynamicKafkaTopics', () => {
+    const validTopics = [
+      { label: 'field1', value: '%{[field]}' },
+      { label: 'field2', value: 'field2' },
+      { label: 'field3', value: '%{[field2]}-%{[field3]}' },
+    ];
+    const invalidBracketTopic = [{ label: '%{[field}', value: '%{[field}' }];
+    const invalidPercentTopic = [{ label: '{[field]}', value: '{[field]}' }];
+    it('should work with valid topics', () => {
+      const res = validateDynamicKafkaTopics(validTopics);
+      expect(res).toBeUndefined();
+    });
+    it("should return error with missing brackets in topic's name", () => {
+      const res = validateDynamicKafkaTopics(invalidBracketTopic);
+      expect(res).toEqual(['Topic should have matching amounts of opening and closing brackets']);
+    });
+    it("should return error with missing percent sign before opening brackets in topic's name", () => {
+      const res = validateDynamicKafkaTopics(invalidPercentTopic);
+      expect(res).toEqual(['Opening brackets should be preceded by a percent sign']);
+    });
+  });
 });

x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/settings/components/edit_output_flyout/output_form_validators.tsx

@@ -19,6 +19,15 @@ const toSecretValidator =
     return validator(value ?? '');
   };
 
+const getAllIndices = (str: string, substring: string): number[] => {
+  const indices = [];
+  let index = str.indexOf(substring);
+  while (index !== -1) {
+    indices.push(index);
+    index = str.indexOf(substring, index + 1);
+  }
+  return indices;
+};
 export function validateKafkaHosts(value: string[]) {
   const res: Array<{ message: string; index?: number }> = [];
   const urlIndexes: { [key: string]: number[] } = {};
@@ -362,12 +371,30 @@ export function validateKafkaStaticTopic(value: string) {
 export function validateDynamicKafkaTopics(value: Array<EuiComboBoxOptionOption<string>>) {
   const res = [];
   value.forEach((val, idx) => {
-    if (!val) {
+    if (!val || !val.value) {
       res.push(
         i18n.translate('xpack.fleet.settings.outputForm.kafkaTopicFieldRequiredMessage', {
           defaultMessage: 'Topic is required',
         })
       );
+    } else {
+      const openingBrackets = getAllIndices(val.value, '{[');
+      const closingBrackets = getAllIndices(val.value, ']}');
+      if (openingBrackets.length !== closingBrackets.length) {
+        res.push(
+          i18n.translate('xpack.fleet.settings.outputForm.kafkaTopicBracketsError', {
+            defaultMessage: 'Topic should have matching amounts of opening and closing brackets',
+          })
+        );
+      }
+      // check for preceding percent sign
+      if (!openingBrackets.every((item) => val?.value![item - 1] === '%')) {
+        res.push(
+          i18n.translate('xpack.fleet.settings.outputForm.kafkaTopicPercentError', {
+            defaultMessage: 'Opening brackets should be preceded by a percent sign',
+          })
+        );
+      }
     }
   });
 
@@ -378,6 +405,7 @@ export function validateDynamicKafkaTopics(value: Array<EuiComboBoxOptionOption<
       })
     );
   }
+
   if (res.length) {
     return res;
   }

x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/settings/components/edit_output_flyout/use_output_form.tsx

@@ -923,7 +923,7 @@ export function useOutputForm(onSucess: () => void, output?: Output, defaultOupu
                   }
                 : kafkaTopicsInput.value === kafkaTopicsType.Dynamic && kafkaDynamicTopicInput.value
                 ? {
-                    topic: `%{[${kafkaDynamicTopicInput.value}]}`,
+                    topic: kafkaDynamicTopicInput.value,
                   }
                 : {}),
               headers: kafkaHeadersInput.value,