23730456890

Committed 30 Mar 2026 06:06AM UTC coverage: 71.003% (+20.5%) from 50.458%

Build # 23730456890

Build Type

push

github

Committed by

web-flow

Commit Message

feat: replace Asset model with native Entity+Edge model (#512)

* feat: replace Asset model with native Entity+Edge model

Replace v1beta2.Asset (rigid proto with typed Any data, separate Owners,
Lineage, Labels fields) with a native Entity+Edge model defined in
raystack/meteor/v1beta1.

Entity has flat properties (structpb.Struct) instead of typed Any wrappers.
All relationships (lineage, ownership) are now first-class Edge objects.
This aligns Meteor with Compass v2's entity-graph model and enables an
open type system where new entity types need no proto changes.

Key changes:
- New proto: raystack/meteor/v1beta1/record.proto (Entity + Edge)
- Record wraps Entity + []Edge instead of *Asset
- All 31 extractors migrated to produce Entity + Edges
- All 3 processors work with Entity.Properties directly
- Compass sink is now a thin Entity->UpsertEntity + Edge->UpsertEdge mapper
- All other sinks (kafka, file, http, stencil, frontier, gcs) updated
- Deleted all 14 typed data schemas (Table, Dashboard, Topic, etc.)
- Fixed structpb.NewStruct compatibility with map[string]string
- Fixed tengo script processor property loss during roundtrip

* chore: remove stale generated files and unused test helpers

- Delete duplicate models/models/raystack/ directory (buf generation artifact)
- Delete test/utils/any.go (BuildAny no longer needed without anypb)
- Delete test/utils/struct.go (BuildStruct no longer needed)

* chore: remove dead TryParseMapToProto and parseMapToProto functions

No production callers remain after the Entity migration.
Test helper replaced with local newStruct() using structpb.NewStruct directly.

* refactor: simplify codebase after Entity model migration

- Delete utils/custom_properties.go — inline AsMap/NewStruct in enrich processor
- Delete knownEntityTypes() whitelist in HTTP extractor — open type system
- Inline getSource() wrapper in stencil sink
- Simplify labels processor to use AsMap/NewStruct pattern (preserves all properties)
- Si... (continued)

Run Details

952 of 1299 new or added lines in 48 files covered. (73.29%)

14 existing lines in 7 files now uncovered.

5889 of 8294 relevant lines covered (71.0%)

0.79 hits per line

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

0.0

/plugins/extractors/github/github.go

package github

import (
        "context"
        _ "embed" // used to print the embedded assets

        "github.com/pkg/errors"

        "github.com/google/go-github/v37/github"
        "github.com/raystack/meteor/models"
        "github.com/raystack/meteor/plugins"
        "github.com/raystack/meteor/registry"
        log "github.com/raystack/salt/observability/logger"
        "golang.org/x/oauth2"
)

//go:embed README.md
var summary string

// Config holds the set of configuration for the extractor
type Config struct {
        Org   string `json:"org" yaml:"org" mapstructure:"org" validate:"required"`
        Token string `json:"token" yaml:"token" mapstructure:"token" validate:"required"`
}

var sampleConfig = `
org: raystack
token: github_token`

var info = plugins.Info{
        Description:  "User list from Github organisation.",
        SampleConfig: sampleConfig,
        Summary:      summary,
        Tags:         []string{"platform", "extractor"},
}

// Extractor manages the extraction of data from the extractor
type Extractor struct {
        plugins.BaseExtractor
        logger log.Logger
        config Config
        client *github.Client
}

// New returns a pointer to an initialized Extractor Object
func New(logger log.Logger) *Extractor {
        e := &Extractor{
                logger: logger,
        }
        e.BaseExtractor = plugins.NewBaseExtractor(info, &e.config)

        return e
}

// Init initializes the extractor
func (e *Extractor) Init(ctx context.Context, config plugins.Config) (err error) {
        if err = e.BaseExtractor.Init(ctx, config); err != nil {
                return err
        }

        ts := oauth2.StaticTokenSource(
                &oauth2.Token{AccessToken: e.config.Token},
        )
        tc := oauth2.NewClient(ctx, ts)
        e.client = github.NewClient(tc)

        return
}

// Extract extracts the data from the extractor
// The data is returned as a list of assets.Asset
func (e *Extractor) Extract(ctx context.Context, emit plugins.Emit) (err error) {
        users, _, err := e.client.Organizations.ListMembers(ctx, e.config.Org, nil)

        if err != nil {
                return errors.Wrap(err, "failed to fetch organizations")
        }
        for _, user := range users {
                usr, _, err := e.client.Users.Get(ctx, *user.Login)
                if err != nil {
                        e.logger.Error("failed to fetch user", "error", err)
                        continue
                }
                props := map[string]interface{}{
                        "email":    usr.GetEmail(),
                        "username": usr.GetLogin(),
                        "full_name": usr.GetName(),
                        "status":   "active",
                }
                entity := models.NewEntity(
                        models.NewURN("github", e.UrnScope, "user", usr.GetNodeID()),
                        "user",
                        "",
                        "github",
                        props,
                )
                emit(models.NewRecord(entity))
        }

        return nil
}

// init registers the extractor to catalog
func init() {
        if err := registry.Extractors.Register("github", func() plugins.Extractor {
                return New(plugins.GetLog())
        }); err != nil {
                panic(err)
        }
}

1	package github
2
3	import (
4	"context"
5	_ "embed" // used to print the embedded assets
6
7	"github.com/pkg/errors"
8
9	"github.com/google/go-github/v37/github"
10	"github.com/raystack/meteor/models"
11	"github.com/raystack/meteor/plugins"
12	"github.com/raystack/meteor/registry"
13	log "github.com/raystack/salt/observability/logger"
14	"golang.org/x/oauth2"
15	)
16
17	//go:embed README.md
18	var summary string
19
20	// Config holds the set of configuration for the extractor
21	type Config struct {
22	Org string `json:"org" yaml:"org" mapstructure:"org" validate:"required"`
23	Token string `json:"token" yaml:"token" mapstructure:"token" validate:"required"`
24	}
25
26	var sampleConfig = `
27	org: raystack
28	token: github_token`
29
30	var info = plugins.Info{
31	Description: "User list from Github organisation.",
32	SampleConfig: sampleConfig,
33	Summary: summary,
34	Tags: []string{"platform", "extractor"},
35	}
36
37	// Extractor manages the extraction of data from the extractor
38	type Extractor struct {
39	plugins.BaseExtractor
40	logger log.Logger
41	config Config
42	client *github.Client
43	}
44
45	// New returns a pointer to an initialized Extractor Object
46	func New(logger log.Logger) *Extractor {	×
47	e := &Extractor{	×
48	logger: logger,	×
49	}	×
50	e.BaseExtractor = plugins.NewBaseExtractor(info, &e.config)	×
51		×
52	return e	×
53	}	×
54
55	// Init initializes the extractor
56	func (e *Extractor) Init(ctx context.Context, config plugins.Config) (err error) {	×
57	if err = e.BaseExtractor.Init(ctx, config); err != nil {	×
58	return err	×
59	}	×
60
61	ts := oauth2.StaticTokenSource(	×
62	&oauth2.Token{AccessToken: e.config.Token},	×
63	)	×
64	tc := oauth2.NewClient(ctx, ts)	×
65	e.client = github.NewClient(tc)	×
66		×
67	return	×
68	}
69
70	// Extract extracts the data from the extractor
71	// The data is returned as a list of assets.Asset
72	func (e *Extractor) Extract(ctx context.Context, emit plugins.Emit) (err error) {	×
73	users, _, err := e.client.Organizations.ListMembers(ctx, e.config.Org, nil)	×
74		×
75	if err != nil {	×
76	return errors.Wrap(err, "failed to fetch organizations")	×
77	}	×
78	for _, user := range users {	×
79	usr, _, err := e.client.Users.Get(ctx, *user.Login)	×
80	if err != nil {	×
81	e.logger.Error("failed to fetch user", "error", err)	×
82	continue	×
83	}
NEW 84	props := map[string]interface{}{	×
NEW 85	"email": usr.GetEmail(),	×
NEW 86	"username": usr.GetLogin(),	×
NEW 87	"full_name": usr.GetName(),	×
NEW 88	"status": "active",	×
89	}	×
NEW 90	entity := models.NewEntity(	×
NEW 91	models.NewURN("github", e.UrnScope, "user", usr.GetNodeID()),	×
NEW 92	"user",	×
NEW 93	"",	×
NEW 94	"github",	×
NEW 95	props,	×
NEW 96	)	×
NEW 97	emit(models.NewRecord(entity))	×
98	}
99
100	return nil	×
101	}
102
103	// init registers the extractor to catalog
104	func init() {	×
105	if err := registry.Extractors.Register("github", func() plugins.Extractor {	×
106	return New(plugins.GetLog())	×
107	}); err != nil {	×
108	panic(err)	×
109	}
110	}

raystack / meteor / 23730456890

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