3673539607

Committed 12 Dec 2022 07:26AM UTC coverage: 74.396% (+0.4%) from 73.973%

Build Type

github

Committed by Suhas Karanth

Commit Message

feat: add HTTP extractor

Pull Request Pull Request #442: feat: add HTTP extractor

Run Details

245 of 245 new or added lines in 5 files covered. (100.0%)

5393 of 7249 relevant lines covered (74.4%)

0.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.2

/plugins/extractors/http/execute_script.go

package http

import (
        "context"
        "errors"
        "fmt"

        "github.com/d5/tengo/v2"
        "github.com/odpf/meteor/models"
        v1beta2 "github.com/odpf/meteor/models/odpf/assets/v1beta2"
        "github.com/odpf/meteor/plugins"
        "github.com/odpf/meteor/plugins/internal/tengoutil"
        "github.com/odpf/meteor/plugins/internal/tengoutil/structmap"
        "google.golang.org/protobuf/proto"
)

func (e *Extractor) executeScript(ctx context.Context, res interface{}, emit plugins.Emit) error {
        s := tengoutil.NewSecureScript(([]byte)(e.config.Script.Source))
        if err := declareGlobals(s, res, emit); err != nil {
                return fmt.Errorf("declare globals: %w", err)
        }

        c, err := s.Compile()
        if err != nil {
                return fmt.Errorf("compile: %w", err)
        }

        if err := c.RunContext(ctx); err != nil && !errors.Is(err, errUserExit) {
                return fmt.Errorf("run: %w", err)
        }

        return nil
}

func declareGlobals(s *tengo.Script, res interface{}, emit plugins.Emit) error {
        for name, v := range map[string]interface{}{
                "response": res,
                "new_asset": &tengo.UserFunction{
                        Name:  "new_asset",
                        Value: newAssetWrapper(),
                },
                "emit": &tengo.UserFunction{
                        Name:  "emit",
                        Value: emitWrapper(emit),
                },
                "exit": &tengo.UserFunction{
                        Name: "exit",
                        Value: func(...tengo.Object) (tengo.Object, error) {
                                return nil, errUserExit
                        },
                },
        } {
                if err := s.Add(name, v); err != nil {
                        return fmt.Errorf("declare script globals: %w", err)
                }
        }
        return nil
}

func newAssetWrapper() tengo.CallableFunc {
        typeURLs := knownTypeURLs()
        return func(args ...tengo.Object) (tengo.Object, error) {
                if len(args) != 1 {
                        return nil, tengo.ErrWrongNumArguments
                }

                typ, ok := tengo.ToString(args[0])
                if !ok {
                        return nil, tengo.ErrInvalidArgumentType{
                                Name:     "typ",
                                Expected: "string(compatible)",
                                Found:    args[0].TypeName(),
                        }
                }

                return newAsset(typeURLs, typ)
        }
}

func emitWrapper(emit plugins.Emit) tengo.CallableFunc {
        return func(args ...tengo.Object) (tengo.Object, error) {
                if len(args) != 1 {
                        return nil, tengo.ErrWrongNumArguments
                }

                m, ok := tengo.ToInterface(args[0]).(map[string]interface{})
                if !ok {
                        return nil, tengo.ErrInvalidArgumentType{
                                Name:     "asset",
                                Expected: "Map",
                                Found:    args[0].TypeName(),
                        }
                }

                var ast v1beta2.Asset
                if err := structmap.AsStruct(m, &ast); err != nil {
                        return nil, fmt.Errorf("emit asset: %w", err)
                }

                emit(models.NewRecord(&ast))

                return tengo.UndefinedValue, nil
        }
}

func newAsset(typeURLs map[string]string, typ string) (tengo.Object, error) {
        u, ok := typeURLs[typ]
        if !ok {
                return nil, fmt.Errorf("new asset: unexpected type: %s", typ)
        }

        return &tengo.Map{
                Value: map[string]tengo.Object{
                        "type": &tengo.String{Value: typ},
                        "data": &tengo.Map{
                                Value: map[string]tengo.Object{
                                        "@type": &tengo.String{Value: u},
                                },
                        },
                },
        }, nil
}

func knownTypeURLs() map[string]string {
        typeURLs := make(map[string]string, 12)
        for _, typ := range []string{
                "bucket", "dashboard", "experiment", "feature_table", "group",
                "job", "metric", "model", "application", "table", "topic", "user",
        } {
                typeURLs[typ] = typeURL(typ)
        }
        return typeURLs
}

func typeURL(typ string) string {
        const prefix = "type.googleapis.com/"

        var msg proto.Message
        switch typ {
        case "bucket":
                msg = &v1beta2.Bucket{}
        case "dashboard":
                msg = &v1beta2.Dashboard{}
        case "experiment":
                msg = &v1beta2.Experiment{}
        case "feature_table":
                msg = &v1beta2.FeatureTable{}
        case "group":
                msg = &v1beta2.Group{}
        case "job":
                msg = &v1beta2.Job{}
        case "metric":
                msg = &v1beta2.Metric{}
        case "model":
                msg = &v1beta2.Model{}
        case "application":
                msg = &v1beta2.Application{}
        case "table":
                msg = &v1beta2.Table{}
        case "topic":
                msg = &v1beta2.Topic{}
        case "user":
                msg = &v1beta2.User{}
        default:
                panic(fmt.Errorf("unexpected type name: %s", typ))
        }

        return prefix + (string)(msg.ProtoReflect().Descriptor().FullName())
}

1	package http
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7
8	"github.com/d5/tengo/v2"
9	"github.com/odpf/meteor/models"
10	v1beta2 "github.com/odpf/meteor/models/odpf/assets/v1beta2"
11	"github.com/odpf/meteor/plugins"
12	"github.com/odpf/meteor/plugins/internal/tengoutil"
13	"github.com/odpf/meteor/plugins/internal/tengoutil/structmap"
14	"google.golang.org/protobuf/proto"
15	)
16
17	func (e *Extractor) executeScript(ctx context.Context, res interface{}, emit plugins.Emit) error {	1✔
18	s := tengoutil.NewSecureScript(([]byte)(e.config.Script.Source))	1✔
19	if err := declareGlobals(s, res, emit); err != nil {	1✔
20	return fmt.Errorf("declare globals: %w", err)	×
21	}	×
22
23	c, err := s.Compile()	1✔
24	if err != nil {	1✔
25	return fmt.Errorf("compile: %w", err)	×
26	}	×
27
28	if err := c.RunContext(ctx); err != nil && !errors.Is(err, errUserExit) {	2✔
29	return fmt.Errorf("run: %w", err)	1✔
30	}	1✔
31
32	return nil	1✔
33	}
34
35	func declareGlobals(s *tengo.Script, res interface{}, emit plugins.Emit) error {	1✔
36	for name, v := range map[string]interface{}{	1✔
37	"response": res,	1✔
38	"new_asset": &tengo.UserFunction{	1✔
39	Name: "new_asset",	1✔
40	Value: newAssetWrapper(),	1✔
41	},	1✔
42	"emit": &tengo.UserFunction{	1✔
43	Name: "emit",	1✔
44	Value: emitWrapper(emit),	1✔
45	},	1✔
46	"exit": &tengo.UserFunction{	1✔
47	Name: "exit",	1✔
48	Value: func(...tengo.Object) (tengo.Object, error) {	2✔
49	return nil, errUserExit	1✔
50	},	1✔
51	},
52	} {	1✔
53	if err := s.Add(name, v); err != nil {	1✔
54	return fmt.Errorf("declare script globals: %w", err)	×
55	}	×
56	}
57	return nil	1✔
58	}
59
60	func newAssetWrapper() tengo.CallableFunc {	1✔
61	typeURLs := knownTypeURLs()	1✔
62	return func(args ...tengo.Object) (tengo.Object, error) {	2✔
63	if len(args) != 1 {	2✔
64	return nil, tengo.ErrWrongNumArguments	1✔
65	}	1✔
66
67	typ, ok := tengo.ToString(args[0])	1✔
68	if !ok {	1✔
69	return nil, tengo.ErrInvalidArgumentType{	×
70	Name: "typ",	×
71	Expected: "string(compatible)",	×
72	Found: args[0].TypeName(),	×
73	}	×
74	}	×
75
76	return newAsset(typeURLs, typ)	1✔
77	}
78	}
79
80	func emitWrapper(emit plugins.Emit) tengo.CallableFunc {	1✔
81	return func(args ...tengo.Object) (tengo.Object, error) {	2✔
82	if len(args) != 1 {	2✔
83	return nil, tengo.ErrWrongNumArguments	1✔
84	}	1✔
85
86	m, ok := tengo.ToInterface(args[0]).(map[string]interface{})	1✔
87	if !ok {	2✔
88	return nil, tengo.ErrInvalidArgumentType{	1✔
89	Name: "asset",	1✔
90	Expected: "Map",	1✔
91	Found: args[0].TypeName(),	1✔
92	}	1✔
93	}	1✔
94
95	var ast v1beta2.Asset	1✔
96	if err := structmap.AsStruct(m, &ast); err != nil {	1✔
97	return nil, fmt.Errorf("emit asset: %w", err)	×
98	}	×
99
100	emit(models.NewRecord(&ast))	1✔
101		1✔
102	return tengo.UndefinedValue, nil	1✔
103	}
104	}
105
106	func newAsset(typeURLs map[string]string, typ string) (tengo.Object, error) {	1✔
107	u, ok := typeURLs[typ]	1✔
108	if !ok {	2✔
109	return nil, fmt.Errorf("new asset: unexpected type: %s", typ)	1✔
110	}	1✔
111
112	return &tengo.Map{	1✔
113	Value: map[string]tengo.Object{	1✔
114	"type": &tengo.String{Value: typ},	1✔
115	"data": &tengo.Map{	1✔
116	Value: map[string]tengo.Object{	1✔
117	"@type": &tengo.String{Value: u},	1✔
118	},	1✔
119	},	1✔
120	},	1✔
121	}, nil	1✔
122	}
123
124	func knownTypeURLs() map[string]string {	1✔
125	typeURLs := make(map[string]string, 12)	1✔
126	for _, typ := range []string{	1✔
127	"bucket", "dashboard", "experiment", "feature_table", "group",	1✔
128	"job", "metric", "model", "application", "table", "topic", "user",	1✔
129	} {	2✔
130	typeURLs[typ] = typeURL(typ)	1✔
131	}	1✔
132	return typeURLs	1✔
133	}
134
135	func typeURL(typ string) string {	1✔
136	const prefix = "type.googleapis.com/"	1✔
137		1✔
138	var msg proto.Message	1✔
139	switch typ {	1✔
140	case "bucket":	1✔
141	msg = &v1beta2.Bucket{}	1✔
142	case "dashboard":	1✔
143	msg = &v1beta2.Dashboard{}	1✔
144	case "experiment":	1✔
145	msg = &v1beta2.Experiment{}	1✔
146	case "feature_table":	1✔
147	msg = &v1beta2.FeatureTable{}	1✔
148	case "group":	1✔
149	msg = &v1beta2.Group{}	1✔
150	case "job":	1✔
151	msg = &v1beta2.Job{}	1✔
152	case "metric":	1✔
153	msg = &v1beta2.Metric{}	1✔
154	case "model":	1✔
155	msg = &v1beta2.Model{}	1✔
156	case "application":	1✔
157	msg = &v1beta2.Application{}	1✔
158	case "table":	1✔
159	msg = &v1beta2.Table{}	1✔
160	case "topic":	1✔
161	msg = &v1beta2.Topic{}	1✔
162	case "user":	1✔
163	msg = &v1beta2.User{}	1✔
164	default:	×
165	panic(fmt.Errorf("unexpected type name: %s", typ))	×
166	}
167
168	return prefix + (string)(msg.ProtoReflect().Descriptor().FullName())	1✔
169	}

odpf / meteor / 3673539607

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous