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
141
142
143
144
145
146
147
|
package grpctool
import (
"context"
"errors"
"fmt"
"net"
"strings"
"github.com/ash2k/stager"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/tool/prototool"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
func HandleIoError(msg string, err error) error {
if IsStatusError(err) {
s := status.Convert(err).Proto()
s.Message = fmt.Sprintf("%s: %s", msg, s.Message)
err = status.ErrorProto(s)
} else {
err = status.Errorf(codes.Canceled, "%s: %v", msg, err)
}
return err
}
func RequestCanceledOrTimedOut(err error) bool {
return RequestCanceled(err) || RequestTimedOut(err)
}
func RequestCanceled(err error) bool {
for err != nil {
if err == context.Canceled { // nolint:errorlint
return true
}
code := status.Code(err)
if code == codes.Canceled {
return true
}
err = errors.Unwrap(err)
}
return false
}
func RequestTimedOut(err error) bool {
for err != nil {
if err == context.DeadlineExceeded { // nolint:errorlint
return true
}
code := status.Code(err)
if code == codes.DeadlineExceeded {
return true
}
err = errors.Unwrap(err)
}
return false
}
func StartServer(stage stager.Stage, server *grpc.Server, listener func() (net.Listener, error), onStop func()) {
stage.Go(func(ctx context.Context) error {
// gRPC listener
lis, err := listener()
if err != nil {
return err
}
return server.Serve(lis)
})
stage.Go(func(ctx context.Context) error {
<-ctx.Done() // can be cancelled because Serve() failed or main ctx was canceled or some stage failed
onStop()
server.GracefulStop()
return nil
})
}
func IsStatusError(err error) bool {
_, ok := err.(interface { // nolint:errorlint
GRPCStatus() *status.Status
})
return ok
}
func MetaToValuesMap(meta metadata.MD) map[string]*prototool.Values {
if len(meta) == 0 {
return nil
}
result := make(map[string]*prototool.Values, len(meta))
for k, v := range meta {
val := make([]string, len(v))
copy(val, v) // metadata may be mutated, so copy
result[k] = &prototool.Values{
Value: val,
}
}
return result
}
func ValuesMapToMeta(vals map[string]*prototool.Values) metadata.MD {
if len(vals) == 0 {
return nil
}
result := make(metadata.MD, len(vals))
keysLen := 0
for _, v := range vals {
keysLen += len(v.Value)
}
keys := make([]string, 0, keysLen) // allocate backing array for all elements in one go
for k, v := range vals {
keys = append(keys, v.Value...)
// set capacity to length to protect against potential append overwriting next value
lk := len(keys)
result[k] = keys[:lk:lk]
keys = keys[lk:]
}
return result
}
func SplitGrpcMethod(fullMethodName string) (string /* service */, string /* method */) {
if fullMethodName != "" && fullMethodName[0] == '/' {
fullMethodName = fullMethodName[1:]
}
pos := strings.LastIndex(fullMethodName, "/")
if pos == -1 {
return "unknown", fullMethodName
}
service := fullMethodName[:pos]
method := fullMethodName[pos+1:]
return service, method
}
// StatusErrorFromContext is a version of status.FromContextError(ctx.Err()).Err() that allows to augment the
// error message.
func StatusErrorFromContext(ctx context.Context, msg string) error {
err := ctx.Err()
var code codes.Code
switch err { // nolint: errorlint
case context.Canceled:
code = codes.Canceled
case context.DeadlineExceeded:
code = codes.DeadlineExceeded
default:
code = codes.Unknown
}
return status.Errorf(code, "%s: %v", msg, err)
}
|
