-
Notifications
You must be signed in to change notification settings - Fork 208
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ENH: allow extension configuration from data prepper configuration (#…
…2851) * ADD: initial implementation on injecting extension config Signed-off-by: George Chen <[email protected]>
- Loading branch information
1 parent
e845966
commit a407e4f
Showing
27 changed files
with
866 additions
and
231 deletions.
There are no files selected for viewing
16 changes: 16 additions & 0 deletions
16
...rc/main/java/org/opensearch/dataprepper/model/annotations/DataPrepperExtensionPlugin.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package org.opensearch.dataprepper.model.annotations; | ||
|
|
||
| import java.lang.annotation.Documented; | ||
| import java.lang.annotation.ElementType; | ||
| import java.lang.annotation.Retention; | ||
| import java.lang.annotation.RetentionPolicy; | ||
| import java.lang.annotation.Target; | ||
|
|
||
| @Documented | ||
| @Retention(RetentionPolicy.RUNTIME) | ||
| @Target({ElementType.TYPE}) | ||
| public @interface DataPrepperExtensionPlugin { | ||
| Class<?> modelType(); | ||
|
|
||
| String rootKey(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
94 changes: 94 additions & 0 deletions
94
...er-core/src/main/java/org/opensearch/dataprepper/parser/PipelinesDataflowModelParser.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package org.opensearch.dataprepper.parser; | ||
|
|
||
| import com.fasterxml.jackson.databind.DeserializationFeature; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; | ||
| import org.opensearch.dataprepper.model.configuration.DataPrepperVersion; | ||
| import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.File; | ||
| import java.io.FileFilter; | ||
| import java.io.FileInputStream; | ||
| import java.io.FileNotFoundException; | ||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.io.SequenceInputStream; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.Stream; | ||
|
|
||
| import static java.lang.String.format; | ||
|
|
||
| public class PipelinesDataflowModelParser { | ||
| private static final Logger LOG = LoggerFactory.getLogger(PipelinesDataflowModelParser.class); | ||
| private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(new YAMLFactory()) | ||
| .enable(DeserializationFeature.FAIL_ON_READING_DUP_TREE_KEY); | ||
|
|
||
| private final String pipelineConfigurationFileLocation; | ||
|
|
||
| public PipelinesDataflowModelParser(final String pipelineConfigurationFileLocation) { | ||
| this.pipelineConfigurationFileLocation = pipelineConfigurationFileLocation; | ||
| } | ||
|
|
||
| public PipelinesDataFlowModel parseConfiguration() { | ||
| try (final InputStream mergedPipelineConfigurationFiles = mergePipelineConfigurationFiles()) { | ||
| final PipelinesDataFlowModel pipelinesDataFlowModel = OBJECT_MAPPER.readValue(mergedPipelineConfigurationFiles, | ||
| PipelinesDataFlowModel.class); | ||
|
|
||
| final DataPrepperVersion version = pipelinesDataFlowModel.getDataPrepperVersion(); | ||
| validateDataPrepperVersion(version); | ||
|
|
||
| return pipelinesDataFlowModel; | ||
| } catch (IOException e) { | ||
| LOG.error("Failed to parse the configuration file {}", pipelineConfigurationFileLocation); | ||
| throw new ParseException(format("Failed to parse the configuration file %s", pipelineConfigurationFileLocation), e); | ||
| } | ||
| } | ||
|
|
||
| private void validateDataPrepperVersion(final DataPrepperVersion version) { | ||
| if (Objects.nonNull(version) && !DataPrepperVersion.getCurrentVersion().compatibleWith(version)) { | ||
| LOG.error("The version: {} is not compatible with the current version: {}", version, DataPrepperVersion.getCurrentVersion()); | ||
| throw new ParseException(format("The version: %s is not compatible with the current version: %s", | ||
| version, DataPrepperVersion.getCurrentVersion())); | ||
| } | ||
| } | ||
|
|
||
| private InputStream mergePipelineConfigurationFiles() throws IOException { | ||
| final File configurationLocation = new File(pipelineConfigurationFileLocation); | ||
|
|
||
| if (configurationLocation.isFile()) { | ||
| return new FileInputStream(configurationLocation); | ||
| } else if (configurationLocation.isDirectory()) { | ||
| FileFilter yamlFilter = pathname -> (pathname.getName().endsWith(".yaml") || pathname.getName().endsWith(".yml")); | ||
| List<InputStream> configurationFiles = Stream.of(configurationLocation.listFiles(yamlFilter)) | ||
| .map(file -> { | ||
| InputStream inputStream; | ||
| try { | ||
| inputStream = new FileInputStream(file); | ||
| LOG.info("Reading pipeline configuration from {}", file.getName()); | ||
| } catch (FileNotFoundException e) { | ||
| inputStream = null; | ||
| LOG.warn("Pipeline configuration file {} not found", file.getName()); | ||
| } | ||
| return inputStream; | ||
| }) | ||
| .filter(Objects::nonNull) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| if (configurationFiles.isEmpty()) { | ||
| LOG.error("Pipelines configuration file not found at {}", pipelineConfigurationFileLocation); | ||
| throw new ParseException( | ||
| format("Pipelines configuration file not found at %s", pipelineConfigurationFileLocation)); | ||
| } | ||
|
|
||
| return new SequenceInputStream(Collections.enumeration(configurationFiles)); | ||
| } else { | ||
| LOG.error("Pipelines configuration file not found at {}", pipelineConfigurationFileLocation); | ||
| throw new ParseException(format("Pipelines configuration file not found at %s", pipelineConfigurationFileLocation)); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.