1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
package main
import (
"context"
"fmt"
"net"
"net/http"
"net/url"
"os"
"os/signal"
"strings"
"time"
"github.com/alecthomas/kong"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/reflection"
castorev1pb "code.tvl.fyi/tvix/castore-go"
"code.tvl.fyi/tvix/nar-bridge/pkg/pathinfosvc"
storev1pb "code.tvl.fyi/tvix/store-go"
"github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
)
// `help:"Provide a tvix-store gRPC PathInfoService for a HTTP Nix Binary Cache"`
var cli struct {
LogLevel string `enum:"trace,debug,info,warn,error,fatal,panic" help:"The log level to log with" default:"info"`
ListenAddr string `name:"listen-addr" help:"The address this service listens on" type:"string" default:"[::]:8001"` //nolint:lll
BlobServiceAddr string `name:"blob-service-addr" env:"BLOB_SERVICE_ADDR" default:"grpc+http://[::1]:8000"`
DirectoryServiceAddr string `name:"directory-service-addr" env:"DIRECTORY_SERVICE_ADDR" default:"grpc+http://[::1]:8000"`
HTTPBinaryCacheURL *url.URL `name:"http-binary-cache-url" env:"HTTP_BINARY_CACHE_URL" help:"The URL containing the Nix HTTP Binary cache" default:"https://cache.nixos.org"`
}
func connectService(ctx context.Context, serviceAddr string) (*grpc.ClientConn, error) {
if !strings.HasPrefix(serviceAddr, "grpc+http://") {
return nil, fmt.Errorf("invalid serviceAddr: %s", serviceAddr)
}
addr := strings.TrimPrefix(serviceAddr, "grpc+http://")
conn, err := grpc.DialContext(ctx, addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
log.Fatalf("did not connect: %v", err)
}
return conn, nil
}
func main() {
_ = kong.Parse(&cli)
logLevel, err := logrus.ParseLevel(cli.LogLevel)
if err != nil {
log.Fatal("invalid log level")
}
logrus.SetLevel(logLevel)
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()
// connect to the two stores
connBlobSvc, err := connectService(ctx, cli.BlobServiceAddr)
if err != nil {
log.Fatalf("unable to connect to blob service: %v", err)
}
defer connBlobSvc.Close()
connDirectorySvc, err := connectService(ctx, cli.DirectoryServiceAddr)
if err != nil {
log.Fatalf("unable to connect to directory service: %v", err)
}
defer connDirectorySvc.Close()
// set up pathinfoservice
var opts []grpc.ServerOption
s := grpc.NewServer(opts...)
reflection.Register(s)
storev1pb.RegisterPathInfoServiceServer(s,
pathinfosvc.New(
cli.HTTPBinaryCacheURL,
&http.Client{},
castorev1pb.NewDirectoryServiceClient(connDirectorySvc),
castorev1pb.NewBlobServiceClient(connBlobSvc),
),
)
log.Printf("Starting nar-bridge-pathinfosvc at %v", cli.ListenAddr)
lis, err := net.Listen("tcp", cli.ListenAddr)
if err != nil {
log.Fatalf("failed to listen: %v", err)
}
go s.Serve(lis)
// listen for the interrupt signal.
<-ctx.Done()
// Restore default behaviour on the interrupt signal
stop()
log.Info("Received Signal, shutting down, press Ctl+C again to force.")
stopped := make(chan interface{})
go func() {
s.GracefulStop()
close(stopped)
}()
t := time.NewTimer(30 * time.Second)
select {
case <-t.C:
log.Info("timeout, kicking remaining clients")
s.Stop()
case <-stopped:
log.Info("all clients left during grace period")
t.Stop()
}
}
|
