[Fleet] Improve validation for dynamic Kafka topics (#212422)

Closes #206194 


## Summary

- Removed hardcoded wrapping of user-entered topics with `%{[]}` to fix
issues arising from the user pre-wrapping, and also allow greater
flexibility in naming
- Added validation rules to check for unclosed brackets & brackets with
missing `%` preceding
- Added the auto-wrapping to the `value` field of items chosen from the
dropdown to ensure they were always wrapped as intended
### Checklist

Check the PR satisfies following conditions. 

Reviewers should verify this PR satisfies this list as well.

- [ ] Any text added follows [EUI's writing
guidelines](https://elastic.github.io/eui/#/guidelines/writing), uses
sentence case text and includes [i18n
support](https://github.com/elastic/kibana/blob/main/src/platform/packages/shared/kbn-i18n/README.md)
- [ ]
[Documentation](https://www.elastic.co/guide/en/kibana/master/development-documentation.html)
was added for features that require explanation or tutorials
- [ ] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios
- [ ] If a plugin configuration key changed, check if it needs to be
allowlisted in the cloud and added to the [docker
list](https://github.com/elastic/kibana/blob/main/src/dev/build/tasks/os_packages/docker_generator/resources/base/bin/kibana-docker)
- [ ] This was checked for breaking HTTP API changes, and any breaking
changes have been approved by the breaking-change committee. The
`release_note:breaking` label should be applied in these situations.
- [ ] [Flaky Test
Runner](https://ci-stats.kibana.dev/trigger_flaky_test_runner/1) was
used on any tests changed
- [ ] The PR description includes the appropriate Release Notes section,
and the correct `release_note:*` label is applied per the
[guidelines](https://www.elastic.co/guide/en/kibana/master/contributing.html#kibana-release-notes-process)

### Identify risks

n/a

---------

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

src/platform/packages/shared/kbn-doc-links/src/get_doc_links.ts

@@ -108,6 +108,7 @@ export const getDocLinks = ({ kibanaBranch, buildFlavor }: GetDocLinkOptions): D
       configuration: `${ELASTIC_WEBSITE_URL}guide/en/beats/filebeat/${DOC_LINK_VERSION}/configuring-howto-filebeat.html`,
       elasticsearchModule: `${ELASTIC_WEBSITE_URL}guide/en/beats/filebeat/${DOC_LINK_VERSION}/filebeat-module-elasticsearch.html`,
       elasticsearchOutput: `${ELASTIC_WEBSITE_URL}guide/en/beats/filebeat/${DOC_LINK_VERSION}/elasticsearch-output.html`,
+      kafkaOutput: `${ELASTIC_WEBSITE_URL}guide/en/beats/filebeat/${DOC_LINK_VERSION}/kafka-output.html`,
       startup: `${ELASTIC_WEBSITE_URL}guide/en/beats/filebeat/${DOC_LINK_VERSION}/filebeat-starting.html`,
       exportedFields: `${ELASTIC_WEBSITE_URL}guide/en/beats/filebeat/${DOC_LINK_VERSION}/exported-fields.html`,
       suricataModule: `${ELASTIC_WEBSITE_URL}guide/en/beats/filebeat/${DOC_LINK_VERSION}/filebeat-module-suricata.html`,

src/platform/packages/shared/kbn-doc-links/src/types.ts

@@ -70,6 +70,7 @@ export interface DocLinks {
     readonly configuration: string;
     readonly elasticsearchOutput: string;
     readonly elasticsearchModule: string;
+    readonly kafkaOutput: string;
     readonly startup: string;
     readonly exportedFields: string;
     readonly suricataModule: string;

x-pack/platform/plugins/private/translations/translations/fr-FR.json

@@ -19015,7 +19015,6 @@
     "xpack.fleet.settings.editOutputFlyout.kafkaCompressionTitle": "Compression",
     "xpack.fleet.settings.editOutputFlyout.kafkaConnectionTypeLabel": "Connexion",
     "xpack.fleet.settings.editOutputFlyout.kafkaDynamicTopicHelptext": "Sélectionnez un sujet dans la liste. Si un sujet n'est pas disponible, créez un sujet personnalisé.",
-    "xpack.fleet.settings.editOutputFlyout.kafkaDynamicTopicLabel": "Sujet du champ",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeaderKeyInputLabel": "Clé",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeadersTitle": "En-têtes",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeaderValueInputLabel": "Valeur",

x-pack/platform/plugins/private/translations/translations/ja-JP.json

@@ -18876,7 +18876,6 @@
     "xpack.fleet.settings.editOutputFlyout.kafkaCompressionTitle": "圧縮",
     "xpack.fleet.settings.editOutputFlyout.kafkaConnectionTypeLabel": "接続",
     "xpack.fleet.settings.editOutputFlyout.kafkaDynamicTopicHelptext": "リストからトピックを選択してください。トピックがない場合は、カスタムトピックを作成してください。",
-    "xpack.fleet.settings.editOutputFlyout.kafkaDynamicTopicLabel": "フィールドからのトピック",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeaderKeyInputLabel": "キー",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeadersTitle": "ヘッダー",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeaderValueInputLabel": "値",

x-pack/platform/plugins/private/translations/translations/zh-CN.json

@@ -18575,7 +18575,6 @@
     "xpack.fleet.settings.editOutputFlyout.kafkaCompressionTitle": "压缩",
     "xpack.fleet.settings.editOutputFlyout.kafkaConnectionTypeLabel": "连接",
     "xpack.fleet.settings.editOutputFlyout.kafkaDynamicTopicHelptext": "从列表中选择主题。如果主题不可用，请创建定制主题。",
-    "xpack.fleet.settings.editOutputFlyout.kafkaDynamicTopicLabel": "来自字段的主题",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeaderKeyInputLabel": "钥匙",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeadersTitle": "标题",
     "xpack.fleet.settings.editOutputFlyout.kafkaHeaderValueInputLabel": "值",

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

@@ -15,10 +15,13 @@ import {
   EuiTitle,
   EuiRadioGroup,
   EuiComboBox,
+  EuiLink,
 } from '@elastic/eui';
 import { FormattedMessage } from '@kbn/i18n-react';
 import { i18n } from '@kbn/i18n';
 
+import { useStartServices } from '../../../../hooks';
+
 import {
   kafkaTopicsType,
   KAFKA_DYNAMIC_FIELDS,
@@ -30,10 +33,12 @@ import type { OutputFormInputsType } from './use_output_form';
 export const OutputFormKafkaTopics: React.FunctionComponent<{ inputs: OutputFormInputsType }> = ({
   inputs,
 }) => {
+  const { docLinks } = useStartServices();
+
   const dynamicOptions: Array<EuiComboBoxOptionOption<string>> = useMemo(() => {
     const options = KAFKA_DYNAMIC_FIELDS.map((option) => ({
       label: option,
-      value: option,
+      value: `%{[${option}]}`,
     }));
     return options;
   }, []);
@@ -73,7 +78,17 @@ 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={docLinks.links.filebeat.kafkaOutput} 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,28 @@ 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([
+        'The topic should have a matching number 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,31 @@ 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:
+              'The topic should have a matching number 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 +406,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,