knotmirror: request index on refUpdate · tangled.org/core@cd7c4ae

+5

knotmirror/config/config.go

··· 16 16 KnotSSRF bool `env:"MIRROR_KNOT_SSRF, default=false"` 17 17 GitRepoBasePath string `env:"MIRROR_GIT_BASEPATH, default=repos"` 18 18 GitRepoFetchTimeout time.Duration `env:"MIRROR_GIT_FETCH_TIMEOUT, default=600s"` 19 + Search SearchConfig `env:",prefix=MIRROR_SEARCH_"` 19 20 ResyncParallelism int `env:"MIRROR_RESYNC_PARALLELISM, default=5"` 20 21 Slurper SlurperConfig `env:",prefix=MIRROR_SLURPER_"` 21 22 UseSSL bool `env:"MIRROR_USE_SSL, default=false"` ··· 35 36 type SlurperConfig struct { 36 37 PersistCursorPeriod time.Duration `env:"PERSIST_CURSOR_PERIOD, default=4s"` 37 38 ConcurrencyPerHost int `env:"CONCURRENCY, default=4"` 39 + } 40 + 41 + type SearchConfig struct { 42 + ZoektUrl string `env:"ZOEKT_URL"` // base url to zoekt node. skipped when empty 38 43 } 39 44 40 45 func Load(ctx context.Context) (*Config, error) {

+48

knotmirror/git.go

··· 17 17 "tangled.org/core/knotmirror/models" 18 18 ) 19 19 20 + type branch struct { 21 + Name string `json:"name"` 22 + Version string `json:"version"` 23 + } 24 + 20 25 type GitMirrorManager interface { 21 26 Exist(repo *models.Repo) (bool, error) 22 27 // Clone clones the repository as a mirror ··· 25 30 Fetch(ctx context.Context, repo *models.Repo) error 26 31 // Sync mirrors the repository. It will clone the repository if repository doesn't exist. 27 32 Sync(ctx context.Context, repo *models.Repo) error 33 + DefaultBranch(ctx context.Context, repo *models.Repo) (branch, error) 28 34 Delete(repo *models.Repo) error 29 35 } 30 36 ··· 149 155 return nil 150 156 } 151 157 158 + func (c *CliGitMirrorManager) DefaultBranch(ctx context.Context, repo *models.Repo) (branch, error) { 159 + path := c.makeRepoPath(repo) 160 + 161 + nameCmd := exec.CommandContext(ctx, "git", "-C", path, "symbolic-ref", "--short", "HEAD") 162 + nameOut, err := nameCmd.Output() 163 + if err != nil { 164 + return branch{}, err 165 + } 166 + 167 + // --verify --quiet exits 1 with no output on an empty repo (unborn HEAD). 168 + revCmd := exec.CommandContext(ctx, "git", "-C", path, "rev-parse", "--verify", "--quiet", "HEAD") 169 + revOut, err := revCmd.Output() 170 + if err != nil { 171 + return branch{}, err 172 + } 173 + 174 + version := strings.TrimSpace(string(revOut)) 175 + if version == "" { 176 + return branch{}, errors.New("git: no commits") 177 + } 178 + 179 + return branch{ 180 + Name: strings.TrimSpace(string(nameOut)), 181 + Version: version, 182 + }, nil 183 + } 184 + 152 185 func (c *CliGitMirrorManager) Delete(repo *models.Repo) error { 153 186 return os.RemoveAll(c.makeRepoPath(repo)) 154 187 } ··· 286 319 } 287 320 } 288 321 return nil 322 + } 323 + 324 + func (c *GoGitMirrorManager) DefaultBranch(ctx context.Context, repo *models.Repo) (branch, error) { 325 + gr, err := git.PlainOpen(c.makeRepoPath(repo)) 326 + if err != nil { 327 + return branch{}, fmt.Errorf("opening local repo: %w", err) 328 + } 329 + ref, err := gr.Head() 330 + if err != nil { 331 + return branch{}, fmt.Errorf("resolving HEAD: %w", err) 332 + } 333 + return branch{ 334 + Name: ref.Name().Short(), 335 + Version: ref.Hash().String(), 336 + }, nil 289 337 } 290 338 291 339 func (c *GoGitMirrorManager) Delete(repo *models.Repo) error {

+48

knotmirror/resyncer.go

··· 4 4 "bytes" 5 5 "context" 6 6 "database/sql" 7 + "encoding/json" 7 8 "errors" 8 9 "fmt" 9 10 "io" ··· 258 259 return false, err 259 260 } 260 261 262 + // request index to zoekt server 263 + // NOTE: indexing after full git resync is bad design. We are doing _after_ the sync because knotstream event doesn't include repository refs. 264 + // NOTE: and zoekt indexer should directly subscribe to the knot. remove this when we have knotrelay. 265 + if r.cfg.Search.ZoektUrl != "" { 266 + go func() { 267 + idxCtx, cancel := context.WithTimeout(ctx, 10*time.Second) 268 + defer cancel() 269 + defaultBranch, err := r.gitm.DefaultBranch(idxCtx, repo) 270 + if err != nil { 271 + r.logger.Warn("resolving default branch for indexing failed", "did", repo.RepoDid, "error", err) 272 + return 273 + } 274 + if err := r.requestIndex(idxCtx, repo.RepoDid, []branch{defaultBranch}); err != nil { 275 + r.logger.Warn("requesting zoekt index failed", "did", repo.RepoDid, "err", err) 276 + } 277 + }() 278 + } 279 + 261 280 // queue repo_stats_update job 262 281 r.indexer.AddTask(context.TODO(), &knotstream.Task{Key: repo.RepoDid.String()}) 263 282 ··· 396 415 jitter := time.Millisecond * time.Duration(rand.Intn(1000)) 397 416 return time.Second*time.Duration(dur) + jitter 398 417 } 418 + 419 + func (r *Resyncer) requestIndex(ctx context.Context, repoDid syntax.DID, branches []branch) error { 420 + r.logger.Info("requesting index", "repo", repoDid, "branches", branches) 421 + body, err := json.Marshal(map[string]any{ 422 + "repo": repoDid.String(), 423 + "branches": branches, 424 + }) 425 + if err != nil { 426 + return fmt.Errorf("marshaling index request: %w", err) 427 + } 428 + 429 + endpoint := r.cfg.Search.ZoektUrl + "/indexserver/admin/enqueueIndex" 430 + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) 431 + if err != nil { 432 + return err 433 + } 434 + req.Header.Set("Content-Type", "application/json") 435 + 436 + resp, err := r.httpClient.Do(req) 437 + if err != nil { 438 + return fmt.Errorf("requesting zoekt index: %w", err) 439 + } 440 + defer resp.Body.Close() 441 + 442 + if resp.StatusCode < 200 || resp.StatusCode >= 300 { 443 + return fmt.Errorf("non-ok status: %d", resp.StatusCode) 444 + } 445 + return nil 446 + }

Configure Feed

Configure Feed