FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

insights: add durable repo iterator (#42676) · TiO2/sourcegraph@ee22fa5 · GitHub

/ sourcegraph Public
forked from sourcegraph/sourcegraph-public-snapshot

Commit ee22fa5

Browse files
coury-clark
authored
insights: add durable repo iterator (sourcegraph#42676)
1 parent 66d332b commit ee22fa5

9 files changed

Lines changed: 869 additions & 0 deletions

File tree

‎enterprise/internal/insights/scheduler/backfill.go‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package scheduler
2+
3+
import "github.com/sourcegraph/sourcegraph/enterprise/internal/insights/types"
4+
5+
type BackfillScheduler interface {
6+
Backfill(series types.InsightSeries) error
7+
}

‎enterprise/internal/insights/scheduler/iterator/repo_iterator.go‎

Lines changed: 291 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,291 @@
1+
package iterator
2+
3+
import (
4+
"context"
5+
"time"
6+
7+
"github.com/derision-test/glock"
8+
"github.com/keegancsmith/sqlf"
9+
"github.com/lib/pq"
10+
11+
"github.com/sourcegraph/sourcegraph/internal/api"
12+
"github.com/sourcegraph/sourcegraph/internal/database/basestore"
13+
"github.com/sourcegraph/sourcegraph/internal/database/dbutil"
14+
"github.com/sourcegraph/sourcegraph/lib/errors"
15+
)
16+
17+
type finishFunc func(ctx context.Context, store *basestore.Store, maybeErr error) error
18+
19+
// persistentRepoIterator represents a durable (persisted) iterator over a set of repositories. This iteration is not
20+
// concurrency safe and only one consumer should have access to this resource at a time.
21+
type persistentRepoIterator struct {
22+
id int
23+
CreatedAt time.Time
24+
StartedAt time.Time
25+
CompletedAt time.Time
26+
RuntimeDuration time.Duration
27+
PercentComplete float64
28+
TotalCount int
29+
SuccessCount int
30+
repos []int32
31+
Cursor int
32+
errors errorMap
33+
34+
itrStart time.Time // time the current iteration started
35+
itrEnd time.Time // time the current iteration ended
36+
37+
glock glock.Clock
38+
}
39+
40+
type errorMap map[int32]*IterationError
41+
42+
type IterationError struct {
43+
id int
44+
RepoId int32
45+
FailureCount int
46+
ErrorMessages []string
47+
}
48+
49+
var repoIteratorCols = []*sqlf.Query{
50+
sqlf.Sprintf("repo_iterator.id"),
51+
sqlf.Sprintf("repo_iterator.created_at"),
52+
sqlf.Sprintf("repo_iterator.started_at"),
53+
sqlf.Sprintf("repo_iterator.completed_at"),
54+
sqlf.Sprintf("repo_iterator.runtime_duration"),
55+
sqlf.Sprintf("repo_iterator.percent_complete"),
56+
sqlf.Sprintf("repo_iterator.total_count"),
57+
sqlf.Sprintf("repo_iterator.success_count"),
58+
sqlf.Sprintf("repo_iterator.repos"),
59+
sqlf.Sprintf("repo_iterator.repo_cursor"),
60+
}
61+
var iteratorJoinCols = sqlf.Join(repoIteratorCols, ", ")
62+
63+
var repoIteratorErrorCols = []*sqlf.Query{
64+
sqlf.Sprintf("repo_iterator_errors.id"),
65+
sqlf.Sprintf("repo_iterator_errors.repo_id"),
66+
sqlf.Sprintf("repo_iterator_errors.error_message"),
67+
sqlf.Sprintf("repo_iterator_errors.failure_count"),
68+
}
69+
var errorJoinCols = sqlf.Join(repoIteratorErrorCols, ", ")
70+
71+
// New returns a new (durable) repo iterator starting from cursor position 0.
72+
func New(ctx context.Context, store *basestore.Store, repos []int32) (*persistentRepoIterator, error) {
73+
return NewWithClock(ctx, store, glock.NewRealClock(), repos)
74+
}
75+
76+
// NewWithClock returns a new (durable) repo iterator starting from cursor position 0 and optionally overrides the internal clock. Useful for tests.
77+
func NewWithClock(ctx context.Context, store *basestore.Store, clock glock.Clock, repos []int32) (*persistentRepoIterator, error) {
78+
if len(repos) == 0 {
79+
return nil, errors.New("unable to construct a repo iterator for an empty set")
80+
}
81+
82+
q := "INSERT INTO repo_iterator(repos, total_count, created_at) VALUES (%S, %S, %S) RETURNING id"
83+
id, err := basestore.ScanInt(store.QueryRow(ctx, sqlf.Sprintf(q, pq.Int32Array(repos), len(repos), clock.Now())))
84+
if err != nil {
85+
return nil, err
86+
}
87+
88+
loaded, err := Load(ctx, store, id)
89+
if err != nil {
90+
return nil, err
91+
}
92+
loaded.glock = clock
93+
return loaded, nil
94+
}
95+
96+
// Load will load a repo iterator that has been persisted and prepare it at the current cursor state.
97+
func Load(ctx context.Context, store *basestore.Store, id int) (got *persistentRepoIterator, err error) {
98+
return LoadWithClock(ctx, store, id, glock.NewRealClock())
99+
}
100+
101+
func LoadWithClock(ctx context.Context, store *basestore.Store, id int, clock glock.Clock) (_ *persistentRepoIterator, err error) {
102+
baseQuery := "SELECT %S FROM repo_iterator WHERE repo_iterator.id = %S"
103+
row := store.QueryRow(ctx, sqlf.Sprintf(baseQuery, iteratorJoinCols, id))
104+
var repos pq.Int32Array
105+
var tmp persistentRepoIterator
106+
if err = row.Scan(
107+
&tmp.id,
108+
&tmp.CreatedAt,
109+
&dbutil.NullTime{Time: &tmp.StartedAt},
110+
&dbutil.NullTime{Time: &tmp.CompletedAt},
111+
&tmp.RuntimeDuration,
112+
&tmp.PercentComplete,
113+
&tmp.TotalCount,
114+
&tmp.SuccessCount,
115+
&repos,
116+
&tmp.Cursor,
117+
); err != nil {
118+
return nil, errors.Wrap(err, "ScanRepoIterator")
119+
}
120+
tmp.repos = repos
121+
if tmp.Cursor > len(tmp.repos) {
122+
return nil, errors.Newf("invalid repo iterator state id:%d cursor:%d length:%d", tmp.id, tmp.Cursor, len(repos))
123+
}
124+
125+
tmp.errors, err = loadRepoIteratorErrors(ctx, store, &tmp)
126+
if err != nil {
127+
return nil, errors.Wrap(err, "loadRepoIteratorErrors")
128+
}
129+
130+
tmp.glock = clock
131+
return &tmp, nil
132+
}
133+
134+
// NextWithFinish will iterate the repository set from the current cursor position. If the iterator is marked complete
135+
// or has no more repositories this will do nothing. The finish function returned is a mechanism to have atomic updates,
136+
// callers will need to call the finish function when complete with work. Errors during work processing can be passed
137+
// into the finish function and will be marked as errors on the repo iterator. Calling NextWithFinish without calling the
138+
// finish function will infinitely loop on the current cursor. This iteration for a given repo iterator is not
139+
// concurrency safe and should only be called from a single thread. Care should be taken to ensure in a distributed
140+
// environment only one consumer is able to access this resource at a time.
141+
func (p *persistentRepoIterator) NextWithFinish() (api.RepoID, bool, finishFunc) {
142+
current, got := p.peek(p.Cursor)
143+
if !p.CompletedAt.IsZero() || !got {
144+
return 0, false, func(ctx context.Context, store *basestore.Store, err error) error {
145+
return nil
146+
}
147+
}
148+
p.itrStart = p.glock.Now()
149+
return api.RepoID(current), true, func(ctx context.Context, store *basestore.Store, maybeErr error) error {
150+
p.itrEnd = p.glock.Now()
151+
if err := p.doFinish(ctx, store, maybeErr, current); err != nil {
152+
return err
153+
}
154+
return nil
155+
}
156+
}
157+
158+
// MarkComplete will mark the repo iterator as complete. Once marked complete the iterator is no longer eligible for iteration.
159+
// This can be called at any time to mark the iterator as complete, and does not require the cursor have passed all the way through the set.
160+
func (p *persistentRepoIterator) MarkComplete(ctx context.Context, store *basestore.Store) error {
161+
now := p.glock.Now()
162+
err := store.Exec(ctx, sqlf.Sprintf("UPDATE repo_iterator SET percent_complete = 1, completed_at = %S, last_updated_at = %S", now, now))
163+
if err != nil {
164+
return err
165+
}
166+
p.CompletedAt = now
167+
p.PercentComplete = 1
168+
return nil
169+
}
170+
171+
func stampStartedAt(ctx context.Context, store *basestore.Store, itrId int, stampTime time.Time) error {
172+
return store.Exec(ctx, sqlf.Sprintf("UPDATE repo_iterator SET started_at = %S WHERE id = %S", stampTime, itrId))
173+
}
174+
175+
func (p *persistentRepoIterator) peek(offset int) (int32, bool) {
176+
if offset >= len(p.repos) {
177+
return 0, false
178+
}
179+
return p.repos[offset], true
180+
}
181+
182+
func (p *persistentRepoIterator) insertIterationError(ctx context.Context, store *basestore.Store, repoId int32, msg string) (err error) {
183+
var query *sqlf.Query
184+
if p.id == 0 {
185+
return errors.New("invalid iterator to insert iterator error")
186+
}
187+
188+
v, ok := p.errors[repoId]
189+
if !ok {
190+
query = sqlf.Sprintf("INSERT INTO repo_iterator_errors(repo_iterator_id, repo_id, error_message) VALUES (%S, %S, %S) RETURNING %S", p.id, repoId, pq.Array([]string{msg}), errorJoinCols)
191+
row := store.QueryRow(ctx, query)
192+
var tmp IterationError
193+
if err = row.Scan(
194+
&tmp.id,
195+
&tmp.RepoId,
196+
pq.Array(&tmp.ErrorMessages),
197+
&tmp.FailureCount,
198+
); err != nil {
199+
return errors.Wrap(err, "InsertIterationError")
200+
}
201+
p.errors[tmp.RepoId] = &tmp
202+
} else {
203+
v.FailureCount += 1
204+
query = sqlf.Sprintf("UPDATE repo_iterator_errors SET failure_count = %S, error_message = array_append(error_message, %S) WHERE id = %S", v.FailureCount, msg, v.id)
205+
if err = store.Exec(ctx, query); err != nil {
206+
return errors.Wrap(err, "UpdateIterationError")
207+
}
208+
}
209+
return nil
210+
}
211+
212+
func (p *persistentRepoIterator) doFinish(ctx context.Context, store *basestore.Store, maybeErr error, cursorVal int32) (err error) {
213+
didSucceed := 0
214+
didAttempt := 1
215+
if maybeErr == nil {
216+
didSucceed = 1
217+
}
218+
itrDuration := p.itrEnd.Sub(p.itrStart)
219+
220+
tx, err := store.Transact(ctx)
221+
if err != nil {
222+
return err
223+
}
224+
defer func() { err = tx.Done(err) }()
225+
226+
updateQ := `UPDATE repo_iterator
227+
SET percent_complete = COALESCE((%s / NULLIF(total_count, 0)), 0),
228+
success_count = success_count + %s,
229+
repo_cursor = repo_cursor + %s,
230+
last_updated_at = NOW(),
231+
runtime_duration = runtime_duration + %s
232+
WHERE id = %s RETURNING percent_complete, success_count, repo_cursor, runtime_duration;`
233+
234+
var pct float64
235+
var successCnt int
236+
var cursor int
237+
var runtime time.Duration
238+
q := sqlf.Sprintf(updateQ, didSucceed, didSucceed, didAttempt, itrDuration, p.id)
239+
row := tx.QueryRow(ctx, q)
240+
if err = row.Scan(
241+
&pct,
242+
&successCnt,
243+
&cursor,
244+
&runtime,
245+
); err != nil {
246+
return errors.Wrapf(err, "unable to update cursor on iteration success iteratorId: %d, new_cursor:%d", p.id, cursor)
247+
}
248+
if maybeErr != nil {
249+
if err = p.insertIterationError(ctx, tx, cursorVal, maybeErr.Error()); err != nil {
250+
return errors.Wrapf(err, "unable to upsert error iteratorId: %d, new_cursor:%d", p.id, cursor)
251+
}
252+
}
253+
if p.StartedAt.IsZero() {
254+
if err = stampStartedAt(ctx, tx, p.id, p.itrStart); err != nil {
255+
return errors.Wrap(err, "stampStartedAt")
256+
}
257+
p.StartedAt = p.itrStart
258+
}
259+
260+
p.Cursor = cursor
261+
p.SuccessCount = successCnt
262+
p.PercentComplete = pct
263+
p.RuntimeDuration = runtime
264+
p.itrStart = time.Time{}
265+
p.itrEnd = time.Time{}
266+
267+
return nil
268+
}
269+
270+
func loadRepoIteratorErrors(ctx context.Context, store *basestore.Store, iterator *persistentRepoIterator) (got errorMap, err error) {
271+
baseQuery := "SELECT %S FROM repo_iterator_errors WHERE repo_iterator_id = %S"
272+
rows, err := store.Query(ctx, sqlf.Sprintf(baseQuery, errorJoinCols, iterator.id))
273+
if err != nil {
274+
return nil, err
275+
}
276+
got = make(errorMap)
277+
for rows.Next() {
278+
var tmp IterationError
279+
if err := rows.Scan(
280+
&tmp.id,
281+
&tmp.RepoId,
282+
pq.Array(&tmp.ErrorMessages),
283+
&tmp.FailureCount,
284+
); err != nil {
285+
return nil, err
286+
}
287+
got[tmp.RepoId] = &tmp
288+
}
289+
290+
return got, err
291+
}

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL