add create/edit endpoints

x-pack/plugins/ingest_pipelines/server/routes/api/create.ts

@@ -0,0 +1,81 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+import { i18n } from '@kbn/i18n';
+import { schema } from '@kbn/config-schema';
+
+import { Pipeline } from '../../../common/types';
+import { API_BASE_PATH } from '../../../common/constants';
+import { RouteDependencies } from '../../types';
+
+const bodySchema = schema.object({
+  name: schema.string(),
+  description: schema.string(),
+  processors: schema.arrayOf(schema.any()), // todo fix
+  version: schema.maybe(schema.number()),
+});
+
+export const registerCreateRoute = ({
+  router,
+  license,
+  lib: { isEsError },
+}: RouteDependencies): void => {
+  router.put(
+    {
+      path: API_BASE_PATH,
+      validate: {
+        body: bodySchema,
+      },
+    },
+    license.guardApiRoute(async (ctx, req, res) => {
+      const { callAsCurrentUser } = ctx.core.elasticsearch.dataClient;
+      const pipeline = req.body as Pipeline;
+
+      const { name, description, processors, version } = pipeline;
+
+      try {
+        // Check that a pipeline with the same name doesn't already exist
+        const pipelineByName = await callAsCurrentUser('ingest.getPipeline', { id: name });
+
+        if (pipelineByName[name]) {
+          return res.conflict({
+            body: new Error(
+              i18n.translate('xpack.ingestPipelines.createRoute.duplicatePipelineIdErrorMessage', {
+                defaultMessage: "There is already a pipeline with name '{name}'.",
+                values: {
+                  name,
+                },
+              })
+            ),
+          });
+        }
+      } catch (e) {
+        // Silently swallow error
+      }
+
+      try {
+        const response = await callAsCurrentUser('ingest.putPipeline', {
+          id: name,
+          body: {
+            description,
+            processors,
+            version,
+          },
+        });
+
+        return res.ok({ body: response });
+      } catch (error) {
+        if (isEsError(error)) {
+          return res.customError({
+            statusCode: error.statusCode,
+            body: error,
+          });
+        }
+
+        return res.internalError({ body: error });
+      }
+    })
+  );
+};

x-pack/plugins/ingest_pipelines/server/routes/api/index.ts

@@ -5,3 +5,7 @@
  */
 
 export { registerGetRoutes } from './get';
+
+export { registerCreateRoute } from './create';
+
+export { registerUpdateRoute } from './update';

x-pack/plugins/ingest_pipelines/server/routes/api/update.ts

@@ -0,0 +1,68 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+import { schema } from '@kbn/config-schema';
+
+import { Pipeline } from '../../../common/types';
+import { API_BASE_PATH } from '../../../common/constants';
+import { RouteDependencies } from '../../types';
+
+const bodySchema = schema.object({
+  description: schema.string(),
+  processors: schema.arrayOf(schema.any()), // todo fix
+  version: schema.maybe(schema.number()),
+});
+
+const paramsSchema = schema.object({
+  name: schema.string(),
+});
+
+export const registerUpdateRoute = ({
+  router,
+  license,
+  lib: { isEsError },
+}: RouteDependencies): void => {
+  router.put(
+    {
+      path: `${API_BASE_PATH}/{name}`,
+      validate: {
+        body: bodySchema,
+        params: paramsSchema,
+      },
+    },
+    license.guardApiRoute(async (ctx, req, res) => {
+      const { callAsCurrentUser } = ctx.core.elasticsearch.dataClient;
+      const { name } = req.params as typeof paramsSchema.type;
+      const pipeline = req.body as Pipeline;
+
+      const { description, processors, version } = pipeline;
+
+      try {
+        // Verify pipeline exists
+        await callAsCurrentUser('ingest.getPipeline', { id: name });
+
+        const response = await callAsCurrentUser('ingest.putPipeline', {
+          id: name,
+          body: {
+            description,
+            processors,
+            version,
+          },
+        });
+
+        return res.ok({ body: response });
+      } catch (error) {
+        if (isEsError(error)) {
+          return res.customError({
+            statusCode: error.statusCode,
+            body: error,
+          });
+        }
+
+        return res.internalError({ body: error });
+      }
+    })
+  );
+};

x-pack/plugins/ingest_pipelines/server/routes/index.ts

@@ -6,10 +6,12 @@
 
 import { RouteDependencies } from '../types';
 
-import { registerGetRoutes } from './api';
+import { registerGetRoutes, registerCreateRoute, registerUpdateRoute } from './api';
 
 export class ApiRoutes {
   setup(dependencies: RouteDependencies) {
     registerGetRoutes(dependencies);
+    registerCreateRoute(dependencies);
+    registerUpdateRoute(dependencies);
   }
 }