Updates to the winning submission for enrich processor

README.md

@@ -37,6 +37,9 @@ Configuration for the application is at config/default.js and config/production.
 - UBAHN_AGGREGATE_TOPIC: Kafka topic that is used to combine all create, update and delete message(s)
 - ES.HOST: Elasticsearch host
 - ES.DOCUMENTS: Elasticsearch index, type and id mapping for resources.
+- ATTRIBUTE_GROUP_PIPELINE_ID: The pipeline id for enrichment with attribute group. Default is `attributegroup-pipeline`
+- SKILL_PROVIDER_PIPELINE_ID: The pipeline id for enrichment with skill provider. Default is `skillprovider-pipeline`
+- USER_PIPELINE_ID: The pipeline id for enrichment of user details. Default is `user-pipeline`
 
 For `ES.DOCUMENTS` configuration, you will find multiple other configurations below it. Each has default values that you can override using the environment variables
 

config/default.js

@@ -51,7 +51,6 @@ module.exports = {
   // ElasticSearch
   ES: {
     HOST: process.env.ES_HOST || 'http://localhost:9200',
-    ENRICH_USER_PIPELINE_NAME: process.env.ENRICH_USER_PIPELINE_NAME || 'enrich_user',
     // es mapping: _index, _type, _id
     DOCUMENTS: {
       achievementprovider: {
@@ -64,7 +63,8 @@ module.exports = {
       },
       attributegroup: {
         index: process.env.ATTRIBUTE_GROUP_INDEX || 'attribute_group',
-        type: '_doc'
+        type: '_doc',
+        pipelineId: process.env.ATTRIBUTE_GROUP_PIPELINE_ID || 'attributegroup-pipeline'
       },
       organization: {
         index: process.env.ORGANIZATION_INDEX || 'organization',
@@ -80,11 +80,13 @@ module.exports = {
       },
       skillprovider: {
         index: process.env.SKILL_PROVIDER_INDEX || 'skill_provider',
-        type: '_doc'
+        type: '_doc',
+        pipelineId: process.env.SKILL_PROVIDER_PIPELINE_ID || 'skillprovider-pipeline'
       },
       user: {
         index: process.env.USER_INDEX || 'user',
-        type: '_doc'
+        type: '_doc',
+        pipelineId: process.env.USER_PIPELINE_ID || 'user-pipeline'
       },
       // sub resources under user
       achievement: {

scripts/constants.js

@@ -6,42 +6,136 @@
 const config = require('config')
 
 const topResources = {
+  skillprovider: {
+    index: config.get('ES.DOCUMENTS.skillprovider.index'),
+    type: config.get('ES.DOCUMENTS.skillprovider.type'),
+    enrich: {
+      policyName: 'skillprovider-policy',
+      matchField: 'id',
+      enrichFields: ['id', 'name', 'created', 'updated', 'createdBy', 'updatedBy']
+    },
+    pipeline: {
+      id: config.get('ES.DOCUMENTS.skillprovider.pipelineId'),
+      field: 'skillProviderId',
+      targetField: 'skillprovider',
+      maxMatches: '1'
+    }
+  },
+
+  role: {
+    index: config.get('ES.DOCUMENTS.role.index'),
+    type: config.get('ES.DOCUMENTS.role.type'),
+    enrich: {
+      policyName: 'role-policy',
+      matchField: 'id',
+      enrichFields: ['id', 'name', 'created', 'updated', 'createdBy', 'updatedBy']
+    }
+  },
+
   achievementprovider: {
     index: config.get('ES.DOCUMENTS.achievementprovider.index'),
-    enrichPolicy: 'achievementprovider-policy',
-    type: config.get('ES.DOCUMENTS.achievementprovider.type')
-  },
-  attribute: {
-    index: config.get('ES.DOCUMENTS.attribute.index'),
-    enrichPolicy: 'attribute-policy',
-    type: config.get('ES.DOCUMENTS.attribute.type')
+    type: config.get('ES.DOCUMENTS.achievementprovider.type'),
+    enrich: {
+      policyName: 'achievementprovider-policy',
+      matchField: 'id',
+      enrichFields: ['id', 'name', 'created', 'updated', 'createdBy', 'updatedBy']
+    }
   },
+
   attributegroup: {
     index: config.get('ES.DOCUMENTS.attributegroup.index'),
-    type: config.get('ES.DOCUMENTS.attributegroup.type')
-  },
-  organization: {
-    index: config.get('ES.DOCUMENTS.organization.index'),
-    enrichPolicy: 'organization-policy',
-    type: config.get('ES.DOCUMENTS.organization.type')
-  },
-  role: {
-    index: config.get('ES.DOCUMENTS.role.index'),
-    enrichPolicy: 'role-policy',
-    type: config.get('ES.DOCUMENTS.role.type')
+    type: config.get('ES.DOCUMENTS.attributegroup.type'),
+    enrich: {
+      policyName: 'attributegroup-policy',
+      matchField: 'id',
+      enrichFields: ['id', 'name', 'organizationId', 'created', 'updated', 'createdBy', 'updatedBy']
+    },
+    pipeline: {
+      id: config.get('ES.DOCUMENTS.attributegroup.pipelineId'),
+      field: 'attributeGroupId',
+      targetField: 'attributegroup',
+      maxMatches: '1'
+    }
   },
+
   skill: {
     index: config.get('ES.DOCUMENTS.skill.index'),
-    enrichPolicy: 'skill-policy',
-    type: config.get('ES.DOCUMENTS.skill.type')
+    type: config.get('ES.DOCUMENTS.skill.type'),
+    enrich: {
+      policyName: 'skill-policy',
+      matchField: 'id',
+      enrichFields: ['id', 'skillProviderId', 'name', 'externalId', 'uri', 'created', 'updated', 'createdBy', 'updatedBy', 'skillprovider']
+    },
+    ingest: {
+      pipeline: {
+        id: config.get('ES.DOCUMENTS.skillprovider.pipelineId')
+      }
+    }
   },
-  skillprovider: {
-    index: config.get('ES.DOCUMENTS.skillprovider.index'),
-    type: config.get('ES.DOCUMENTS.skillprovider.type')
+
+  attribute: {
+    index: config.get('ES.DOCUMENTS.attribute.index'),
+    type: config.get('ES.DOCUMENTS.attribute.type'),
+    enrich: {
+      policyName: 'attribute-policy',
+      matchField: 'id',
+      enrichFields: ['id', 'name', 'attributeGroupId', 'created', 'updated', 'createdBy', 'updatedBy', 'attributegroup']
+    },
+    ingest: {
+      pipeline: {
+        id: config.get('ES.DOCUMENTS.attributegroup.pipelineId')
+      }
+    }
   },
+
+  organization: {
+    index: config.get('ES.DOCUMENTS.organization.index'),
+    type: config.get('ES.DOCUMENTS.organization.type')
+  },
+
   user: {
     index: config.get('ES.DOCUMENTS.user.index'),
-    type: config.get('ES.DOCUMENTS.user.type')
+    type: config.get('ES.DOCUMENTS.user.type'),
+    pipeline: {
+      id: config.get('ES.DOCUMENTS.user.pipelineId'),
+      processors: [
+        {
+          referenceField: config.get('ES.DOCUMENTS.achievement.userField'),
+          enrichPolicyName: 'achievementprovider-policy',
+          field: '_ingest._value.achievementsProviderId',
+          targetField: '_ingest._value.achievementprovider',
+          maxMatches: '1'
+        },
+        {
+          referenceField: config.get('ES.DOCUMENTS.externalprofile.userField'),
+          enrichPolicyName: 'organization-policy',
+          field: '_ingest._value.organizationId',
+          targetField: '_ingest._value.organization',
+          maxMatches: '1'
+        },
+        {
+          referenceField: config.get('ES.DOCUMENTS.userattribute.userField'),
+          enrichPolicyName: 'attribute-policy',
+          field: '_ingest._value.attributeId',
+          targetField: '_ingest._value.attribute',
+          maxMatches: '1'
+        },
+        {
+          referenceField: config.get('ES.DOCUMENTS.userrole.userField'),
+          enrichPolicyName: 'role-policy',
+          field: '_ingest._value.roleId',
+          targetField: '_ingest._value.role',
+          maxMatches: '1'
+        },
+        {
+          referenceField: config.get('ES.DOCUMENTS.userskill.userField'),
+          enrichPolicyName: 'skill-policy',
+          field: '_ingest._value.skillId',
+          targetField: '_ingest._value.skill',
+          maxMatches: '1'
+        }
+      ]
+    }
   }
 }
 
@@ -72,7 +166,12 @@ const userResources = {
 const organizationResources = {
   organizationskillprovider: {
     propertyName: config.get('ES.DOCUMENTS.organizationskillprovider.orgField'),
-    relateKey: 'skillProviderId'
+    relateKey: 'skillProviderId',
+    enrich: {
+      policyName: 'organization-policy',
+      matchField: 'id',
+      enrichFields: ['id', 'name', 'created', 'updated', 'createdBy', 'updatedBy', 'skillProviders']
+    }
   }
 }
 

scripts/db/dropAll.js

@@ -4,24 +4,63 @@
 const _ = require('lodash')
 const models = require('../../src/models')
 const logger = require('../../src/common/logger')
-const { topResources, modelToESIndexMapping } = require('../constants')
+const {
+  topResources,
+  organizationResources,
+  modelToESIndexMapping
+} = require('../constants')
 const { getESClient } = require('../../src/common/es-client')
 
 async function main () {
   const client = getESClient()
+
+  try {
+    logger.info('Deleting all pipelines...')
+    await client.ingest.deletePipeline({
+      id: topResources.user.pipeline.id
+    })
+    await client.ingest.deletePipeline({
+      id: topResources.skillprovider.pipeline.id
+    })
+    await client.ingest.deletePipeline({
+      id: topResources.attributegroup.pipeline.id
+    })
+    logger.info('Successfully deleted')
+  } catch (e) {
+    console.error(e)
+    logger.warn('Delete all ingest pipelines failed')
+  }
+
   const keys = Object.keys(models)
   for (let i = 0; i < keys.length; i++) {
     const key = keys[i]
     if (models[key].tableName) {
       const esResourceName = modelToESIndexMapping[key]
       try {
         if (_.includes(_.keys(topResources), esResourceName)) {
+          if (topResources[esResourceName].enrich) {
+            logger.info(`Deleting enrich policy for ${esResourceName}`)
+            await client.enrich.deletePolicy({
+              name: topResources[esResourceName].enrich.policyName
+            })
+            logger.info(`Successfully deleted enrich policy for ${esResourceName}`)
+          }
+          logger.info(`Deleting index for ${esResourceName}`)
           await client.indices.delete({
             index: topResources[esResourceName].index
           })
+          logger.info(`Successfully deleted enrich policy for ${esResourceName}`)
+        } else if (_.includes(_.keys(organizationResources), esResourceName)) {
+          logger.info('Deleting enrich policy for organization')
+          await client.enrich.deletePolicy({
+            name: organizationResources[esResourceName].enrich.policyName
+          })
+          logger.info('Successfully deleted enrich policy for organization')
         }
 
+        logger.info(`Deleting data in QLDB for ${esResourceName}`)
         await models.DBHelper.clear(models[key])
+        logger.info(`Successfully deleted data in QLDB for ${esResourceName}`)
       } catch (e) {
         console.error(e)
         logger.warn(`drop table ${key} failed`)