Skip to content

Commit d5ba47c

Browse files
feat!: modify URN format to follow ODPF standard (#387)
* feat: add base plugin * fix: breaking plugin test * fix: lint errors * feat: create a NewURN function * test: add missing tests * feat: update all extractors to use odpf urn * feat(plugins): add BaseExtractor * fix: breaking test * fix: improve test coverage * fix: config not passed properly
1 parent b536051 commit d5ba47c

107 files changed

Lines changed: 1862 additions & 1772 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

agent/agent.go

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,10 @@ func (r *Agent) Validate(rcp recipe.Recipe) (errs []error) {
6161
if ext, err := r.extractorFactory.Get(rcp.Source.Name); err != nil {
6262
errs = append(errs, err)
6363
} else {
64-
if err = ext.Validate(rcp.Source.Config); err != nil {
64+
if err = ext.Validate(plugins.Config{
65+
URNScope: rcp.Source.Scope,
66+
RawConfig: rcp.Source.Config,
67+
}); err != nil {
6568
errs = append(errs, r.enrichInvalidConfigError(err, rcp.Source.Name, plugins.PluginTypeExtractor))
6669
}
6770
}
@@ -72,7 +75,7 @@ func (r *Agent) Validate(rcp recipe.Recipe) (errs []error) {
7275
errs = append(errs, err)
7376
continue
7477
}
75-
if err = sink.Validate(s.Config); err != nil {
78+
if err = sink.Validate(plugins.Config{RawConfig: s.Config}); err != nil {
7679
errs = append(errs, r.enrichInvalidConfigError(err, s.Name, plugins.PluginTypeSink))
7780
}
7881
}
@@ -83,7 +86,7 @@ func (r *Agent) Validate(rcp recipe.Recipe) (errs []error) {
8386
errs = append(errs, err)
8487
continue
8588
}
86-
if err = procc.Validate(p.Config); err != nil {
89+
if err = procc.Validate(plugins.Config{RawConfig: p.Config}); err != nil {
8790
errs = append(errs, r.enrichInvalidConfigError(err, p.Name, plugins.PluginTypeProcessor))
8891
}
8992
}
@@ -197,7 +200,7 @@ func (r *Agent) setupExtractor(ctx context.Context, sr recipe.PluginRecipe, str
197200
err = errors.Wrapf(err, "could not find extractor \"%s\"", sr.Name)
198201
return
199202
}
200-
if err = extractor.Init(ctx, sr.Config); err != nil {
203+
if err = extractor.Init(ctx, recipeToPluginConfig(sr)); err != nil {
201204
err = errors.Wrapf(err, "could not initiate extractor \"%s\"", sr.Name)
202205
return
203206
}
@@ -217,7 +220,7 @@ func (r *Agent) setupProcessor(ctx context.Context, pr recipe.PluginRecipe, str
217220
if proc, err = r.processorFactory.Get(pr.Name); err != nil {
218221
return errors.Wrapf(err, "could not find processor \"%s\"", pr.Name)
219222
}
220-
if err = proc.Init(ctx, pr.Config); err != nil {
223+
if err = proc.Init(ctx, recipeToPluginConfig(pr)); err != nil {
221224
return errors.Wrapf(err, "could not initiate processor \"%s\"", pr.Name)
222225
}
223226

@@ -240,7 +243,7 @@ func (r *Agent) setupSink(ctx context.Context, sr recipe.PluginRecipe, stream *s
240243
if sink, err = r.sinkFactory.Get(sr.Name); err != nil {
241244
return errors.Wrapf(err, "could not find sink \"%s\"", sr.Name)
242245
}
243-
if err = sink.Init(ctx, sr.Config); err != nil {
246+
if err = sink.Init(ctx, recipeToPluginConfig(sr)); err != nil {
244247
return errors.Wrapf(err, "could not initiate sink \"%s\"", sr.Name)
245248
}
246249
retryNotification := func(e error, d time.Duration) {

0 commit comments

Comments
 (0)