Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Move InvokeGenerator to a centralized function of Meshkit #543

Closed
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
331 changes: 331 additions & 0 deletions generators/registry.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,331 @@
package generators

import (
"context"
"fmt"
"os"
"strconv"
"path/filepath"
"encoding/json"
"net/url"
"sync"
"time"

"google.golang.org/api/sheets/v4"
"github.com/layer5io/meshkit/utils/walker"
"github.com/layer5io/meshkit/models/meshmodel/core/v1beta1"
"github.com/layer5io/meshkit/utils"
"github.com/layer5io/meshkit/generators/github"
"github.com/layer5io/meshkit/utils/store"
"github.com/layer5io/meshkit/utils/registry"
"golang.org/x/sync/semaphore"
"github.com/layer5io/meshkit/logger"
)

var (
srv *sheets.Service
totalAggregateModel int
totalAggregateComponents int
componentSpredsheetGID int64
sheetGID int64
registryLocation string
logFile *os.File
errorLogFile *os.File
Log logger.Handler
LogError logger.Handler
)

var (
artifactHubCount = 0
artifactHubRateLimit = 100
artifactHubRateLimitDur = 5 * time.Minute
artifactHubMutex sync.Mutex
defVersion = "v1.0.0"
GoogleSpreadSheetURL = "https://docs.google.com/spreadsheets/d/"
logDirPath = filepath.Join(utils.GetHome(), ".meshery", "logs", "registry")
modelToCompGenerateTracker = store.NewGenericThreadSafeStore[registry.CompGenerateTracker]()
)

func InvokeGenerationFromSheet(wg *sync.WaitGroup, spreadsheetID string, spreadsheetCred string) error {

weightedSem := semaphore.NewWeighted(20)
url := GoogleSpreadSheetURL + spreadsheetID
totalAvailableModels := 0
spreadsheeetChan := make(chan registry.SpreadsheetData)

defer func() {
logModelGenerationSummary(modelToCompGenerateTracker)

Log.UpdateLogOutput(os.Stdout)
Log.UpdateLogOutput(os.Stdout)
Log.Info(fmt.Sprintf("Summary: %d models, %d components generated.", totalAggregateModel, totalAggregateComponents))

Log.Info("See ", logDirPath, " for detailed logs.")

_ = logFile.Close()
_ = errorLogFile.Close()
totalAggregateModel = 0
totalAggregateComponents = 0
}()

modelCSVHelper, err := parseModelSheet(url)
if err != nil {
return err
}

componentCSVHelper, err := parseComponentSheet(url)
if err != nil {
return err
}

Log.UpdateLogOutput(logFile)
Log.UpdateLogOutput(errorLogFile)
var wgForSpreadsheetUpdate sync.WaitGroup
wgForSpreadsheetUpdate.Add(1)
go func() {
registry.ProcessModelToComponentsMap(componentCSVHelper.Components)
registry.VerifyandUpdateSpreadsheet(spreadsheetCred, &wgForSpreadsheetUpdate, srv, spreadsheeetChan, spreadsheetID)
}()

// Iterate models from the spreadsheet
for _, model := range modelCSVHelper.Models {
totalAvailableModels++

ctx := context.Background()

err := weightedSem.Acquire(ctx, 1)
if err != nil {
break
}

wg.Add(1)
go func(model registry.ModelCSV) {
defer func() {
wg.Done()
weightedSem.Release(1)
}()
if utils.ReplaceSpacesAndConvertToLowercase(model.Registrant) == "meshery" {
err = GenerateDefsForCoreRegistrant(model)
if err != nil {
LogError.Error(err)
}
return
}

generator, err := NewGenerator(model.Registrant, model.SourceURL, model.Model)
if err != nil {
LogError.Error(registry.ErrGenerateModel(err, model.Model))
return
}

if utils.ReplaceSpacesAndConvertToLowercase(model.Registrant) == "artifacthub" {
rateLimitArtifactHub()

}
pkg, err := generator.GetPackage()
if err != nil {
LogError.Error(registry.ErrGenerateModel(err, model.Model))
return
}

version := pkg.GetVersion()
modelDirPath, compDirPath, err := CreateVersionedDirectoryForModelAndComp(version, model.Model)
if err != nil {
LogError.Error(registry.ErrGenerateModel(err, model.Model))
return
}
modelDef, err := writeModelDefToFileSystem(&model, version, modelDirPath)
if err != nil {
LogError.Error(err)
return
}

comps, err := pkg.GenerateComponents()
if err != nil {
LogError.Error(registry.ErrGenerateModel(err, model.Model))
return
}
Log.Info("Current model: ", model.Model)
Log.Info(" extracted ", len(comps), " components for ", model.ModelDisplayName, " (", model.Model, ")")
for _, comp := range comps {
comp.Version = defVersion
if comp.Metadata == nil {
comp.Metadata = make(map[string]interface{})
}
// Assign the component status corresponding to model status.
// i.e. If model is enabled comps are also "enabled". Ultimately all individual comps itself will have ability to control their status.
// The status "enabled" indicates that the component will be registered inside the registry.
comp.Model = *modelDef
assignDefaultsForCompDefs(&comp, modelDef)
err := comp.WriteComponentDefinition(compDirPath)
if err != nil {
Log.Info(err)
}
}

spreadsheeetChan <- registry.SpreadsheetData{
Model: &model,
Components: comps,
}

modelToCompGenerateTracker.Set(model.Model, registry.CompGenerateTracker{
TotalComps: len(comps),
Version: version,
})
}(model)

}
wg.Wait()
close(spreadsheeetChan)
wgForSpreadsheetUpdate.Wait()
return nil
}

func writeModelDefToFileSystem(model *registry.ModelCSV, version, modelDefPath string) (*v1beta1.Model, error) {
modelDef := model.CreateModelDefinition(version, defVersion)
err := modelDef.WriteModelDefinition(modelDefPath+"/model.json", "json")
if err != nil {
return nil, err
}

return &modelDef, nil
}

func rateLimitArtifactHub() {
artifactHubMutex.Lock()
defer artifactHubMutex.Unlock()

if artifactHubCount > 0 && artifactHubCount%artifactHubRateLimit == 0 {
Log.Info("Rate limit reached for Artifact Hub. Sleeping for 5 minutes...")
time.Sleep(artifactHubRateLimitDur)
}
artifactHubCount++
}

func logModelGenerationSummary(modelToCompGenerateTracker *store.GenerticThreadSafeStore[registry.CompGenerateTracker]) {
for key, val := range modelToCompGenerateTracker.GetAllPairs() {
Log.Info(fmt.Sprintf("Generated %d components for model [%s] %s", val.TotalComps, key, val.Version))
totalAggregateComponents += val.TotalComps
totalAggregateModel++
}

Log.Info(fmt.Sprintf("-----------------------------\n-----------------------------\nGenerated %d models and %d components", totalAggregateModel, totalAggregateComponents))
}

func parseModelSheet(url string) (*registry.ModelCSVHelper, error) {
modelCSVHelper, err := registry.NewModelCSVHelper(url, "Models", sheetGID)
if err != nil {
return nil, err
}

err = modelCSVHelper.ParseModelsSheet(false)
if err != nil {
return nil, registry.ErrGenerateModel(err, "unable to start model generation")
}
return modelCSVHelper, nil
}

func parseComponentSheet(url string) (*registry.ComponentCSVHelper, error) {
compCSVHelper, err := registry.NewComponentCSVHelper(url, "Components", componentSpredsheetGID)
if err != nil {
return nil, err
}
err = compCSVHelper.ParseComponentsSheet()
if err != nil {
return nil, registry.ErrGenerateModel(err, "unable to start model generation")
}
return compCSVHelper, nil
}

func CreateVersionedDirectoryForModelAndComp(version, modelName string) (string, string, error) {
modelDirPath := filepath.Join(registryLocation, modelName, version, defVersion)
err := utils.CreateDirectory(modelDirPath)
if err != nil {
return "", "", err
}

compDirPath := filepath.Join(modelDirPath, "components")
err = utils.CreateDirectory(compDirPath)
return modelDirPath, compDirPath, err
}


func GenerateDefsForCoreRegistrant(model registry.ModelCSV) error {
totalComps := 0
var version string
defer func() {
modelToCompGenerateTracker.Set(model.Model, registry.CompGenerateTracker{
TotalComps: totalComps,
Version: version,
})
}()

path, err := url.Parse(model.SourceURL)
if err != nil {
err = registry.ErrGenerateModel(err, model.Model)
LogError.Error(err)
return nil
}
gitRepo := github.GitRepo{
URL: path,
PackageName: model.Model,
}
owner, repo, branch, root, err := gitRepo.ExtractRepoDetailsFromSourceURL()
if err != nil {
err = registry.ErrGenerateModel(err, model.Model)
LogError.Error(err)
return nil
}

isModelPublished, _ := strconv.ParseBool(model.PublishToRegistry)
//Initialize walker
gitWalker := walker.NewGit()
if isModelPublished {
gw := gitWalker.
Owner(owner).
Repo(repo).
Branch(branch).
Root(root).
RegisterFileInterceptor(func(f walker.File) error {
// Check if the file has a JSON extension
if filepath.Ext(f.Name) != ".json" {
return nil
}
contentBytes := []byte(f.Content)
var componentDef v1beta1.ComponentDefinition
if err := json.Unmarshal(contentBytes, &componentDef); err != nil {
return err
}
version = componentDef.Model.Model.Version
modelDirPath, compDirPath, err := CreateVersionedDirectoryForModelAndComp(version, model.Model)
if err != nil {
err = registry.ErrGenerateModel(err, model.Model)
return err
}
_, err = writeModelDefToFileSystem(&model, version, modelDirPath) // how to infer this? @Beginner86 any idea? new column?
if err != nil {
return registry.ErrGenerateModel(err, model.Model)
}

err = componentDef.WriteComponentDefinition(compDirPath)
if err != nil {
err = registry.ErrGenerateComponent(err, model.Model, componentDef.DisplayName)
LogError.Error(err)
}
return nil
})
err = gw.Walk()
if err != nil {
return err
}
}

return nil
}

func assignDefaultsForCompDefs(componentDef *v1beta1.ComponentDefinition, modelDef *v1beta1.Model) {
componentDef.Metadata["status"] = modelDef.Status
for k, v := range modelDef.Metadata {
componentDef.Metadata[k] = v
}
}
3 changes: 2 additions & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ require (
github.com/fluxcd/pkg/tar v0.4.0
github.com/go-git/go-git/v5 v5.11.0
github.com/go-logr/logr v1.3.0
github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1
github.com/gofrs/uuid v4.4.0+incompatible
github.com/google/go-containerregistry v0.17.0
github.com/google/uuid v1.5.0
Expand All @@ -33,6 +34,7 @@ require (
github.com/spf13/cobra v1.8.0
github.com/spf13/viper v1.17.0
golang.org/x/oauth2 v0.15.0
golang.org/x/sync v0.6.0
golang.org/x/text v0.14.0
google.golang.org/api v0.152.0
gopkg.in/yaml.v2 v2.4.0
Expand Down Expand Up @@ -237,7 +239,6 @@ require (
golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
golang.org/x/mod v0.14.0 // indirect
golang.org/x/net v0.23.0 // indirect
golang.org/x/sync v0.6.0 // indirect
golang.org/x/sys v0.18.0 // indirect
golang.org/x/term v0.18.0 // indirect
golang.org/x/time v0.5.0 // indirect
Expand Down
2 changes: 2 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -354,6 +354,8 @@ github.com/gobuffalo/packr/v2 v2.8.3 h1:xE1yzvnO56cUC0sTpKR3DIbxZgB54AftTFMhB2XE
github.com/gobuffalo/packr/v2 v2.8.3/go.mod h1:0SahksCVcx4IMnigTjiFuyldmTrdTctXsOdiU5KwbKc=
github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1 h1:FWNFq4fM1wPfcK40yHE5UO3RUdSNPaBC+j3PokzA6OQ=
github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1/go.mod h1:5YoVOkjYAQumqlV356Hj3xeYh4BdZuLE0/nRkf2NKkI=
github.com/godbus/dbus v0.0.0-20190422162347-ade71ed3457e/go.mod h1:bBOAhwG1umN6/6ZUMtDFBMQR8jRg9O75tm9K00oMsK4=
github.com/gofrs/uuid v4.4.0+incompatible h1:3qXRTX8/NbyulANqlc0lchS1gqAVxRgsuW1YrTJupqA=
github.com/gofrs/uuid v4.4.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
Expand Down
Loading
Loading