Skip to content

Status table to add possibility to use those data for progress chart #48

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.

Sign up for GitHub

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

Merged
se7entyse7en merged 5 commits into from
Jun 21, 2019
Merged

Status table to add possibility to use those data for progress chart #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0fad366
Add creation of status table
se7entyse7en Jun 20, 2019
c376fb1
Add initialization of status table for shallow sub command
se7entyse7en Jun 20, 2019
2c2d8c9
Add progress tracking for repository and user
se7entyse7en Jun 20, 2019
42c2577
Add tracking of failed resources
se7entyse7en Jun 20, 2019
725187e
Rename part column of status table into entity
se7entyse7en Jun 21, 2019
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
39 changes: 39 additions & 0 deletions cmd/ghsync/subcmd/common.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
)

const maxVersion uint = 1560510971
const statusTableName = "status"

type PostgresOpt struct {
DB string `long:"postgres-db" env:"GHSYNC_POSTGRES_DB" description:"PostgreSQL DB" default:"ghsync"`
Expand Down Expand Up @@ -72,9 +73,47 @@ func (o PostgresOpt) initDB() (db *sql.DB, err error) {

log.With(log.Fields{"db-version": dbVersion}).Debugf("the DB version is up to date")
log.Infof("connection with the DB established")
if err = o.createStatusTable(); err != nil {
return db, err
}

return db, nil
}

func (o PostgresOpt) createStatusTable() error {

@smacker smacker Jun 20, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

whoooh. Why don't use kallax for this table as well?

@se7entyse7en se7entyse7en Jun 21, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

whoooh.

😂

BTW, have a look here. I actually wanted to use kallax, but given the deadline and that I'm completely new to kallax we agreed with @carlosms to just use raw sql instead. But I personally +1 on using kallax, please let me know if you consider it as a blocker, otherwise I'll open an issue.

@smacker smacker Jun 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

okay. I tried to add a new model and re-generate code with it using instruction in readme but it didn't work. So just an issue would be fine for now.

@se7entyse7en se7entyse7en Jun 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, just to verify that I understood correctly. So after this PR we won't be able to create new models right? Until we also handle this with kallax.

@smacker smacker Jun 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. I tried to create a model from master. I mean apparently, it's not that easy to change the current code to use kallax so I'm ok with merging it as it is now.

@se7entyse7en se7entyse7en Jun 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok and thanks for giving it a try.

@se7entyse7en se7entyse7en Jun 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Opened issue.

log.Debugf(fmt.Sprintf("creating status table '%s'", statusTableName))

db, err := sql.Open("postgres", o.URL())
if err != nil {
return err
}

defer func() {
if err != nil {
db.Close()
}
}()

stm := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s(
id serial PRIMARY KEY,
org VARCHAR (50) NOT NULL,
entity VARCHAR (20) NOT NULL,
done INTEGER NOT NULL DEFAULT 0,
failed INTEGER NOT NULL DEFAULT 0,
total INTEGER DEFAULT NULL,
UNIQUE (org, entity)
);`, statusTableName)
log.Debugf("running statement: %s", stm)
_, err = db.Exec(stm)
if err != nil {
return fmt.Errorf("an error occured while ensureing the status table: %v", err)
}

log.Infof("status table '%s' created", statusTableName)

return nil
}

func newMigrate(url string) (*migrate.Migrate, error) {
// wrap assets into Resource
s := bindata.Resource(migrations.AssetNames(),
Expand Down
34 changes: 32 additions & 2 deletions cmd/ghsync/subcmd/shallow.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,14 @@
package subcmd

import (
"database/sql"
"fmt"
"strings"

"github.com/src-d/ghsync/shallow"

"gopkg.in/src-d/go-cli.v0"
"gopkg.in/src-d/go-log.v1"
)

type ShallowCommand struct {
Expand All @@ -29,8 +32,13 @@ func (c *ShallowCommand) Execute(args []string) error {
return err
}

orgSyncer := shallow.NewOrganizationSyncer(db, client)
for _, o := range strings.Split(c.Orgs, ",") {
orgs := strings.Split(c.Orgs, ",")
if err = c.initStatus(db, statusTableName, orgs); err != nil {
return err
}

orgSyncer := shallow.NewOrganizationSyncer(db, client, statusTableName)
for _, o := range orgs {
err = orgSyncer.Sync(o)
if err != nil {
return err
Expand All @@ -39,3 +47,25 @@ func (c *ShallowCommand) Execute(args []string) error {

return nil
}

func (c *ShallowCommand) initStatus(db *sql.DB, tableName string, orgs []string) error {
log.Debugf("initializing status table for orgs: %v", orgs)
var b strings.Builder

for _, o := range orgs[:len(orgs)-1] {

@smacker smacker Jun 20, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

whoooh. kallax would make it much better

b.WriteString(fmt.Sprintf("('%s', 'repository'),", o))
b.WriteString(fmt.Sprintf("('%s', 'user'),", o))
}
b.WriteString(fmt.Sprintf("('%s', 'repository'),", orgs[len(orgs)-1]))
b.WriteString(fmt.Sprintf("('%s', 'user')", orgs[len(orgs)-1]))

stm := fmt.Sprintf("INSERT INTO %s (org, entity) VALUES %s ON CONFLICT (org, entity) DO UPDATE SET failed=0, done=0, total=NULL;", tableName, b.String())
log.Debugf("running statement: %s", stm)
_, err := db.Exec(stm)
if err != nil {
return fmt.Errorf(fmt.Sprintf(
"an error occured while initializing %s table: %v", tableName, err))
}

return nil
}
26 changes: 17 additions & 9 deletions shallow/organization.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,18 @@ import (
)

type OrganizationSyncer struct {
db *sql.DB
store *models.OrganizationStore
client *github.Client
db *sql.DB
store *models.OrganizationStore
client *github.Client
statusTableName string
}

func NewOrganizationSyncer(db *sql.DB, c *github.Client) *OrganizationSyncer {
func NewOrganizationSyncer(db *sql.DB, c *github.Client, statusTableName string) *OrganizationSyncer {
return &OrganizationSyncer{
db: db,
store: models.NewOrganizationStore(db),
client: c,
db: db,
store: models.NewOrganizationStore(db),
client: c,
statusTableName: statusTableName,
}
}

Expand All @@ -40,6 +42,12 @@ func (s *OrganizationSyncer) Sync(login string) error {

if err == nil {
logger.Infof("resource already exists, skipping")
stm := fmt.Sprintf("UPDATE %s SET total=0 WHERE org='%s'", s.statusTableName, login)
_, err = s.db.Exec(stm)
if err != nil {
return fmt.Errorf("unable to update status for org %s: %v", login, err)
}

return nil
}

Expand All @@ -48,13 +56,13 @@ func (s *OrganizationSyncer) Sync(login string) error {
return err
}

repoSyncer := NewRepositorySyncer(s.db, s.client)
repoSyncer := NewRepositorySyncer(s.db, s.client, s.statusTableName)
err = repoSyncer.Sync(login, logger)
if err != nil {
return err
}

userSyncer := NewUserSyncer(s.db, s.client)
userSyncer := NewUserSyncer(s.db, s.client, s.statusTableName)
err = userSyncer.Sync(login, logger)
if err != nil {
return err
Expand Down
46 changes: 39 additions & 7 deletions shallow/repository.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,18 @@ import (
)

type RepositorySyncer struct {
db *sql.DB
store *models.RepositoryStore
client *github.Client
db *sql.DB
store *models.RepositoryStore
client *github.Client
statusTableName string
}

func NewRepositorySyncer(db *sql.DB, c *github.Client) *RepositorySyncer {
func NewRepositorySyncer(db *sql.DB, c *github.Client, statusTableName string) *RepositorySyncer {
return &RepositorySyncer{
db: db,
store: models.NewRepositoryStore(db),
client: c,
db: db,
store: models.NewRepositoryStore(db),
client: c,
statusTableName: statusTableName,
}
}

Expand Down Expand Up @@ -52,10 +54,30 @@ func (s *RepositorySyncer) Sync(owner string, logger log.Logger) error {
opts.Page = r.NextPage
}

stm := fmt.Sprintf("UPDATE %s SET total=%d WHERE org='%s' AND entity='repository'",
s.statusTableName, len(repos), owner)
log.Debugf("running statement: %s", stm)
if _, err := s.db.Exec(stm); err != nil {
return fmt.Errorf("an error occured while updating %s table: %v",
s.statusTableName, err)
}

// Process each one of them
for _, repository := range repos {
err := s.doRepo(repository, logger)
if err != nil {
stm := fmt.Sprintf("UPDATE %s SET failed=failed + 1 WHERE org='%s' AND entity='repository'",
s.statusTableName, owner)
if err = s.updateStatus(stm); err != nil {

@dpordomingo dpordomingo Jun 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

imo it would be better if we'd count the success/fail, and then update the status at once when all the repos have been iterated.

@se7entyse7en se7entyse7en Jun 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But you would lose the possibility to have a pseudo-real-time status chart then, don't you? In case of a sync that requires a lot of time, I think that it's really useful to have it instead of having a report of what happened just in the end.

@dpordomingo dpordomingo Jun 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

real time is great, but with current implementation we're stopping the process if one import or status update fails.

@se7entyse7en se7entyse7en Jun 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yup, but if it fails after 1h as I user I could be interested during that hour about the progress. Maybe as a user having 100 of repos once it finishes 50 repos I may want to start doing some charts, and if update just in the end I won't be able to do that.

@smacker smacker Jun 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

imo better keep as it is now and maybe change exiting on an error

@se7entyse7en se7entyse7en Jun 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

imo better keep as it is now and maybe change exiting on an error

yup, I also prefer to keep it as-is because in the short?/mid?/-term IMO we should change the behaviour of exiting at first error.

return err
}

return err
}

stm := fmt.Sprintf("UPDATE %s SET done=done + 1 WHERE org='%s' AND entity='repository'",
s.statusTableName, owner)
if err = s.updateStatus(stm); err != nil {
return err
}
}
Expand All @@ -65,6 +87,16 @@ func (s *RepositorySyncer) Sync(owner string, logger log.Logger) error {
return nil
}

func (s *RepositorySyncer) updateStatus(stm string) error {
log.Debugf("running statement: %s", stm)
if _, err := s.db.Exec(stm); err != nil {
return fmt.Errorf("an error occured while updating %s table: %v",
s.statusTableName, err)
}

return nil
}

func (s *RepositorySyncer) doRepo(repository *github.Repository, parentLogger log.Logger) error {
logger := parentLogger.With(log.Fields{"repository": repository.GetName()})

Expand Down
116 changes: 82 additions & 34 deletions shallow/user.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,20 +13,23 @@ import (
)

type UserSyncer struct {
db *sql.DB
client *github.Client
db *sql.DB
store *models.UserStore
client *github.Client
statusTableName string
}

func NewUserSyncer(db *sql.DB, c *github.Client) *UserSyncer {
func NewUserSyncer(db *sql.DB, c *github.Client, statusTableName string) *UserSyncer {
return &UserSyncer{
db: db,
client: c,
db: db,
store: models.NewUserStore(db),
client: c,
statusTableName: statusTableName,
}
}

func (s *UserSyncer) Sync(org string, logger log.Logger) error {
store := models.NewUserStore(s.db)
return store.Transaction(func(store *models.UserStore) error {
return s.store.Transaction(func(store *models.UserStore) error {
return s.doUsers(store, org, logger)
})
}
Expand All @@ -37,51 +40,96 @@ func (s *UserSyncer) doUsers(store *models.UserStore, org string, logger log.Log

logger.Infof("starting to retrieve users")

allUsers := make([]*github.User, 0)

// Get the list of all users
for {
users, r, err := s.client.Organizations.ListMembers(context.TODO(), org, opts)
if err != nil {
return err
}

for _, user := range users {
logger := logger.With(log.Fields{"user": user.GetLogin()})

_, err := store.FindOne(models.NewUserQuery().
Where(kallax.And(
kallax.Eq(models.Schema.User.ID, user.GetID()),
)),
)
if err != nil && err != kallax.ErrNotFound {
logger.With(log.Fields{"user": user.GetLogin()}).Errorf(err, "failed to read the resource from the DB")
return fmt.Errorf("failed to read the resource from the DB: %v", err)
}
for _, u := range users {
allUsers = append(allUsers, u)
}

if err == nil {
logger.With(log.Fields{"user": user.GetLogin()}).Infof("resource already exists, skipping")
continue
}
if r.NextPage == 0 {
break
}

record := models.NewUser()
record.User = *user
opts.Page = r.NextPage
}

err = store.Insert(record)
if err != nil {
logger.Errorf(err, "failed to write the resource into the DB")
return fmt.Errorf("failed to write the resource into the DB: %v", err)
stm := fmt.Sprintf("UPDATE %s SET total=%d WHERE org='%s' AND entity='user'",
s.statusTableName, len(allUsers), org)
log.Debugf("running statement: %s", stm)
if _, err := s.db.Exec(stm); err != nil {
Comment thread
dpordomingo marked this conversation as resolved.
return fmt.Errorf("an error occured while updating %s table: %v",
s.statusTableName, err)
}

for _, user := range allUsers {
err := s.doUser(user, logger)
if err != nil {
stm := fmt.Sprintf("UPDATE %s SET failed=failed + 1 WHERE org='%s' AND entity='user'",
s.statusTableName, org)
if err = s.updateStatus(stm); err != nil {
return err
}

logger.Debugf("resource written in the DB")
return err
}

if r.NextPage == 0 {
break
stm := fmt.Sprintf("UPDATE %s SET done=done + 1 WHERE org='%s' AND entity='user'",
s.statusTableName, org)
if err = s.updateStatus(stm); err != nil {
return err
}

opts.Page = r.NextPage
}

logger.Infof("finished to retrieve users")

return nil
}

func (s *UserSyncer) updateStatus(stm string) error {
log.Debugf("running statement: %s", stm)
if _, err := s.db.Exec(stm); err != nil {
return fmt.Errorf("an error occured while updating %s table: %v",
s.statusTableName, err)
}

return nil
}

func (s *UserSyncer) doUser(user *github.User, parentLogger log.Logger) error {
logger := parentLogger.With(log.Fields{"user": user.GetLogin()})

_, err := s.store.FindOne(models.NewUserQuery().
Where(kallax.And(
kallax.Eq(models.Schema.User.ID, user.GetID()),
)),
)
if err != nil && err != kallax.ErrNotFound {
logger.With(log.Fields{"user": user.GetLogin()}).Errorf(err, "failed to read the resource from the DB")
return fmt.Errorf("failed to read the resource from the DB: %v", err)
}

if err == nil {
logger.With(log.Fields{"user": user.GetLogin()}).Infof("resource already exists, skipping")
return nil
}

record := models.NewUser()
record.User = *user

err = s.store.Insert(record)
if err != nil {
logger.Errorf(err, "failed to write the resource into the DB")
return fmt.Errorf("failed to write the resource into the DB: %v", err)
}

logger.Debugf("resource written in the DB")

return nil
}