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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
package starter
import (
"errors"
"fmt"
"net"
"os"
"strings"
"github.com/prometheus/client_golang/prometheus"
"github.com/sirupsen/logrus"
"gitlab.com/gitlab-org/gitaly/internal/bootstrap"
)
const (
// TCP is the prefix for tcp
TCP string = "tcp"
// TLS is the prefix for tls
TLS string = "tls"
// Unix is the prefix for unix
Unix string = "unix"
separator = "://"
)
var (
// ErrEmptySchema signals that the address has no schema in it.
ErrEmptySchema = errors.New("empty schema can't be used")
errEmptyAddress = errors.New("empty address can't be used")
)
// ParseEndpoint returns Config based on the passed in address string.
// Returns error only if provided endpoint has no schema or address defined.
func ParseEndpoint(endpoint string) (Config, error) {
if endpoint == "" {
return Config{}, errEmptyAddress
}
parts := strings.Split(endpoint, separator)
if len(parts) != 2 {
return Config{}, fmt.Errorf("unsupported format: %q: %w", endpoint, ErrEmptySchema)
}
if err := verifySchema(parts[0]); err != nil {
return Config{}, err
}
if parts[1] == "" {
return Config{}, errEmptyAddress
}
return Config{Name: parts[0], Addr: parts[1]}, nil
}
// ComposeEndpoint returns address string composed from provided schema and schema-less address.
func ComposeEndpoint(schema, address string) (string, error) {
if address == "" {
return "", errEmptyAddress
}
if err := verifySchema(schema); err != nil {
return "", err
}
return schema + separator + address, nil
}
func verifySchema(schema string) error {
switch schema {
case "":
return ErrEmptySchema
case TCP, TLS, Unix:
return nil
default:
return fmt.Errorf("unsupported schema: %q", schema)
}
}
// Config represents a network type, and address
type Config struct {
Name, Addr string
// HandoverOnUpgrade indicates whether the socket should be handed over to the new
// process during an upgrade. If the socket is not handed over, it should be be unique
// to avoid colliding with the old process' socket. If the socket is a Unix socket, a
// possible existing file at the path is removed.
HandoverOnUpgrade bool
}
// Endpoint returns fully qualified address.
func (c *Config) Endpoint() (string, error) {
return ComposeEndpoint(c.Name, c.Addr)
}
// IsSecure returns true if network is secured.
func (c *Config) IsSecure() bool {
return c.Name == TLS
}
func (c *Config) family() string {
if c.IsSecure() {
return TCP
}
return c.Name
}
// Server able to serve requests.
type Server interface {
// Serve accepts requests from the listener and handles them properly.
Serve(lis net.Listener) error
}
// New creates a new bootstrap.Starter from a config and a GracefulStoppableServer
func New(cfg Config, server Server) bootstrap.Starter {
return func(listenWithHandover bootstrap.ListenFunc, errCh chan<- error, connTotal *prometheus.CounterVec) error {
listen := listenWithHandover
if !cfg.HandoverOnUpgrade {
if cfg.Name == Unix {
if err := os.Remove(cfg.Addr); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("remove previous socket file: %w", err)
}
}
listen = net.Listen
}
l, err := listen(cfg.family(), cfg.Addr)
if err != nil {
return err
}
logrus.WithField("address", cfg.Addr).Infof("listening at %s address", cfg.Name)
l = wrap(cfg.Name, l, connTotal)
go func() {
errCh <- server.Serve(l)
}()
return nil
}
}
|
