optimus/core/job/resolver/internal_upstream_resolver.go at e80e130366cf8d8c201f49e55251e72a4e7e512f · goto/optimus · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package resolver

import (
	"fmt"

	"golang.org/x/net/context"

	"github.com/goto/optimus/core/job"
	"github.com/goto/optimus/core/resource"
	"github.com/goto/optimus/core/tenant"
	"github.com/goto/optimus/internal/errors"
)

type internalUpstreamResolver struct {
	jobRepository JobRepository
}

func NewInternalUpstreamResolver(jobRepository JobRepository) *internalUpstreamResolver {
	return &internalUpstreamResolver{jobRepository: jobRepository}
}

func (i internalUpstreamResolver) Resolve(ctx context.Context, jobWithUnresolvedUpstream *job.WithUpstream) (*job.WithUpstream, error) {
	me := errors.NewMultiError("internal upstream resolution errors")

	internalUpstreamInferred, err := i.resolveInferredUpstream(ctx, jobWithUnresolvedUpstream.Job().Sources())
	me.Append(err)

	var internalUpstreamStatic []*job.Upstream
	if staticUpstreamSpec := jobWithUnresolvedUpstream.Job().Spec().UpstreamSpec(); staticUpstreamSpec != nil {
		internalUpstreamStatic, err = i.resolveStaticUpstream(ctx, jobWithUnresolvedUpstream.Job().Tenant().ProjectName(), staticUpstreamSpec)
		me.Append(err)
	}

	internalUpstream := mergeUpstreams(internalUpstreamInferred, internalUpstreamStatic)
	distinctInternalUpstream := job.Upstreams(internalUpstream).Deduplicate()
	fullNameUpstreamMap := job.Upstreams(distinctInternalUpstream).ToFullNameAndUpstreamMap()
	resourceDestinationUpstreamMap := job.Upstreams(distinctInternalUpstream).ToResourceDestinationAndUpstreamMap()

	var upstreamResults []*job.Upstream
	for _, unresolvedUpstream := range jobWithUnresolvedUpstream.Upstreams() {
		if resolvedUpstream, ok := resourceDestinationUpstreamMap[unresolvedUpstream.Resource().String()]; ok {
			upstreamResults = append(upstreamResults, resolvedUpstream)
			continue
		}
		if resolvedUpstream, ok := fullNameUpstreamMap[unresolvedUpstream.FullName()]; ok {
			upstreamResults = append(upstreamResults, resolvedUpstream)
			continue
		}
		upstreamResults = append(upstreamResults, unresolvedUpstream)
	}

	distinctUpstreams := job.Upstreams(upstreamResults).Deduplicate()
	jobWithMergedUpstream := job.NewWithUpstream(jobWithUnresolvedUpstream.Job(), distinctUpstreams)
	return jobWithMergedUpstream, me.ToErr()
}

func (i internalUpstreamResolver) BulkResolve(ctx context.Context, projectName tenant.ProjectName, jobsWithUnresolvedUpstream []*job.WithUpstream) ([]*job.WithUpstream, error) {
	jobNames := job.WithUpstreams(jobsWithUnresolvedUpstream).GetSubjectJobNames()

	allInternalUpstreamMap, err := i.jobRepository.ResolveUpstreams(ctx, projectName, jobNames)
	if err != nil {
		errorMsg := fmt.Sprintf("unable to resolve upstream: %s", err.Error())
		return nil, errors.NewError(errors.ErrInternalError, job.EntityJob, errorMsg)
	}

	jobsWithMergedUpstream := job.WithUpstreams(jobsWithUnresolvedUpstream).MergeWithResolvedUpstreams(allInternalUpstreamMap)
	return jobsWithMergedUpstream, nil
}

func (i internalUpstreamResolver) resolveInferredUpstream(ctx context.Context, sources []resource.URN) ([]*job.Upstream, error) {
	var internalUpstream []*job.Upstream
	me := errors.NewMultiError("resolve internal inferred upstream errors")
	for _, source := range sources {
		jobUpstreams, err := i.jobRepository.GetAllEnabledByResourceDestination(ctx, source)
		me.Append(err)
		if len(jobUpstreams) == 0 {
			continue
		}
		upstream := job.NewUpstreamResolved(jobUpstreams[0].Spec().Name(), "", jobUpstreams[0].Destination(), jobUpstreams[0].Tenant(), job.UpstreamTypeInferred, jobUpstreams[0].Spec().Task().Name(), false)
		internalUpstream = append(internalUpstream, upstream)
	}
	return internalUpstream, me.ToErr()
}

func (i internalUpstreamResolver) resolveStaticUpstream(ctx context.Context, projectName tenant.ProjectName, upstreamSpec *job.UpstreamSpec) ([]*job.Upstream, error) {
	var internalUpstream []*job.Upstream
	me := errors.NewMultiError("resolve internal static upstream errors")
	for _, upstreamName := range upstreamSpec.UpstreamNames() {
		upstreamJobName, err := upstreamName.GetJobName()
		if err != nil {
			me.Append(err)
			continue
		}
		jobUpstream, err := i.jobRepository.GetEnabledJobByName(ctx, projectName, upstreamJobName)
		if err != nil || jobUpstream == nil {
			me.Append(err)
			continue
		}
		upstream := job.NewUpstreamResolved(upstreamJobName, "", jobUpstream.Destination(), jobUpstream.Tenant(), job.UpstreamTypeStatic, jobUpstream.Spec().Task().Name(), false)
		internalUpstream = append(internalUpstream, upstream)
	}
	return internalUpstream, me.ToErr()
}