Files
kopia/internal/server/api_sources.go
Jarek Kowalski af1550ad81 feat(server): reduce server refreshes of the repository (#3265)
* feat(server): reduce server refreshes of the repository

Previously each source would refresh itself from the repository
very frequently to determine the upcoming snapshot time. This change
refactors source manager so it does not own the repository connection
on its own but instead delegates all policy reads through the server.

Also introduces a new server scheduler that is responsible for
centrally managing the snapshot schedule and triggering snapshots
when they are due.

* Update cli/command_server_start.go

Co-authored-by: Shikhar Mall <mall.shikhar.in@gmail.com>

* Update internal/server/server.go

Co-authored-by: Shikhar Mall <mall.shikhar.in@gmail.com>

* Update internal/server/server_maintenance.go

Co-authored-by: Shikhar Mall <mall.shikhar.in@gmail.com>

* pr feedback

---------

Co-authored-by: Shikhar Mall <mall.shikhar.in@gmail.com>
2023-09-15 12:08:35 -07:00

95 lines
2.5 KiB
Go

package server
import (
"context"
"encoding/json"
"os"
"sort"
"github.com/pkg/errors"
"github.com/kopia/kopia/internal/ospath"
"github.com/kopia/kopia/internal/serverapi"
"github.com/kopia/kopia/repo"
"github.com/kopia/kopia/snapshot"
"github.com/kopia/kopia/snapshot/policy"
)
func handleSourcesList(ctx context.Context, rc requestContext) (interface{}, *apiError) {
_, multiUser := rc.rep.(repo.DirectRepository)
resp := &serverapi.SourcesResponse{
Sources: []*serverapi.SourceStatus{},
LocalHost: rc.rep.ClientOptions().Hostname,
LocalUsername: rc.rep.ClientOptions().Username,
MultiUser: multiUser,
}
for src, v := range rc.srv.snapshotAllSourceManagers() {
if sourceMatchesURLFilter(src, rc.req.URL.Query()) {
resp.Sources = append(resp.Sources, v.Status())
}
}
sort.Slice(resp.Sources, func(i, j int) bool {
return resp.Sources[i].Source.String() < resp.Sources[j].Source.String()
})
return resp, nil
}
func handleSourcesCreate(ctx context.Context, rc requestContext) (interface{}, *apiError) {
var req serverapi.CreateSnapshotSourceRequest
if err := json.Unmarshal(rc.body, &req); err != nil {
return nil, requestError(serverapi.ErrorMalformedRequest, "malformed request body")
}
if req.Path == "" {
return nil, requestError(serverapi.ErrorMalformedRequest, "missing path")
}
if req.Policy == nil {
return nil, requestError(serverapi.ErrorMalformedRequest, "missing policy")
}
req.Path = ospath.ResolveUserFriendlyPath(req.Path, true)
_, err := os.Stat(req.Path)
if os.IsNotExist(err) {
return nil, requestError(serverapi.ErrorPathNotFound, "path does not exist")
}
if err != nil {
return nil, internalServerError(err)
}
sourceInfo := snapshot.SourceInfo{
UserName: rc.rep.ClientOptions().Username,
Host: rc.rep.ClientOptions().Hostname,
Path: req.Path,
}
resp := &serverapi.CreateSnapshotSourceResponse{}
if err = repo.WriteSession(ctx, rc.rep, repo.WriteSessionOptions{
Purpose: "handleSourcesCreate",
}, func(ctx context.Context, w repo.RepositoryWriter) error {
//nolint:wrapcheck
return policy.SetPolicy(ctx, w, sourceInfo, req.Policy)
}); err != nil {
return nil, internalServerError(errors.Wrap(err, "unable to set initial policy"))
}
manager := rc.srv.getOrCreateSourceManager(ctx, sourceInfo)
if req.CreateSnapshot {
resp.SnapshotStarted = true
log(ctx).Debugf("scheduling snapshot of %v immediately...", sourceInfo)
manager.scheduleSnapshotNow()
}
return resp, nil
}