Skip to content

Commit 9025df3

Browse files
qiujian16qiujian16
committed
Add tests and some refactor
Signed-off-by: Jian Qiu <jqiu@redhat.com>
1 parent 0585f1b commit 9025df3

14 files changed

Lines changed: 794 additions & 25 deletions

File tree

‎pkg/cloudevents/generic/options/grpc/agentoptions.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ func (o *grpcAgentOptions) Protocol(ctx context.Context, dataType types.CloudEve
4343
}),
4444
}
4545
if o.ServerHealthinessTimeout != nil {
46-
opts = append(opts, protocol.WithReconnectErrorOption(o.errorChan, *o.ServerHealthinessTimeout))
46+
opts = append(opts, protocol.WithHealthCheck(*o.ServerHealthinessTimeout, o.errorChan))
4747
}
4848

4949
receiver, err := o.GetCloudEventsProtocol(

‎pkg/cloudevents/generic/options/grpc/options_test.go‎

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
3636
name: "customized options",
3737
config: "{\"url\":\"test\"}",
3838
expectedOptions: &GRPCOptions{
39-
&GRPCDialer{
39+
Dialer: &GRPCDialer{
4040
URL: "test",
4141
KeepAliveOptions: KeepAliveOptions{
4242
Enable: false,
@@ -45,13 +45,14 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
4545
PermitWithoutStream: false,
4646
},
4747
},
48+
ServerHealthinessTimeout: nil,
4849
},
4950
},
5051
{
5152
name: "customized options with yaml format",
5253
config: "url: test",
5354
expectedOptions: &GRPCOptions{
54-
&GRPCDialer{
55+
Dialer: &GRPCDialer{
5556
URL: "test",
5657
KeepAliveOptions: KeepAliveOptions{
5758
Enable: false,
@@ -60,13 +61,14 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
6061
PermitWithoutStream: false,
6162
},
6263
},
64+
ServerHealthinessTimeout: nil,
6365
},
6466
},
6567
{
6668
name: "customized options with keepalive",
67-
config: "{\"url\":\"test\",\"keepAliveConfig\":{\"enable\":true,\"time\":10s,\"timeout\":5s,\"permitWithoutStream\":true}}",
69+
config: "{\"url\":\"test\",\"keepAliveConfig\":{\"enable\":true,\"time\":\"10s\",\"timeout\":\"5s\",\"permitWithoutStream\":true}}",
6870
expectedOptions: &GRPCOptions{
69-
&GRPCDialer{
71+
Dialer: &GRPCDialer{
7072
URL: "test",
7173
KeepAliveOptions: KeepAliveOptions{
7274
Enable: true,
@@ -75,6 +77,23 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
7577
PermitWithoutStream: true,
7678
},
7779
},
80+
ServerHealthinessTimeout: nil,
81+
},
82+
},
83+
{
84+
name: "customized options with server healthiness timeout",
85+
config: "{\"url\":\"test\",\"serverHealthinessTimeout\":\"1m\"}",
86+
expectedOptions: &GRPCOptions{
87+
Dialer: &GRPCDialer{
88+
URL: "test",
89+
KeepAliveOptions: KeepAliveOptions{
90+
Enable: false,
91+
Time: 30 * time.Second,
92+
Timeout: 10 * time.Second,
93+
PermitWithoutStream: false,
94+
},
95+
},
96+
ServerHealthinessTimeout: func() *time.Duration { d := time.Minute; return &d }(),
7897
},
7998
},
8099
}
@@ -85,7 +104,7 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
85104
if err != nil {
86105
t.Fatal(err)
87106
}
88-
defer os.Remove(file.Name())
107+
t.Cleanup(func() { _ = os.Remove(file.Name()) })
89108

90109
options, err := BuildGRPCOptionsFromFlags(file.Name())
91110
if err != nil {

‎pkg/cloudevents/generic/options/grpc/protocol/option.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,12 +26,12 @@ func WithSubscribeOption(subscribeOpt *SubscribeOption) Option {
2626
}
2727
}
2828

29-
func WithReconnectErrorOption(reconnectError chan error, interval time.Duration) Option {
29+
func WithHealthCheck(interval time.Duration, errorChan chan error) Option {
3030
return func(p *Protocol) error {
31-
if reconnectError == nil {
31+
if errorChan == nil {
3232
return fmt.Errorf("the reconnect error option must not be nil")
3333
}
34-
p.reconnectErrorChan = reconnectError
34+
p.reconnectErrorChan = errorChan
3535
if interval <= 0 {
3636
p.checkInterval = 20 * time.Second
3737
} else {

‎pkg/cloudevents/generic/options/grpc/protocol/option_test.go‎

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
package protocol
2+
3+
import (
4+
"context"
5+
"google.golang.org/grpc/credentials/insecure"
6+
"net"
7+
"testing"
8+
"time"
9+
10+
"google.golang.org/grpc"
11+
"google.golang.org/grpc/test/bufconn"
12+
)
13+
14+
func TestWithSubscribeOption(t *testing.T) {
15+
lis := bufconn.Listen(1024)
16+
defer lis.Close()
17+
18+
s := grpc.NewServer()
19+
defer s.Stop()
20+
21+
conn, err := grpc.NewClient("passthrough:///bufnet",
22+
grpc.WithTransportCredentials(insecure.NewCredentials()),
23+
grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
24+
return lis.Dial()
25+
}))
26+
if err != nil {
27+
t.Fatalf("Failed to create connection: %v", err)
28+
}
29+
defer conn.Close()
30+
31+
// Wait for connection to be ready
32+
time.Sleep(50 * time.Millisecond)
33+
34+
// Test valid subscribe option
35+
subscribeOpt := &SubscribeOption{
36+
Source: "test-source",
37+
ClusterName: "test-cluster",
38+
DataType: "test-type",
39+
}
40+
41+
p, err := NewProtocol(conn, WithSubscribeOption(subscribeOpt))
42+
if err != nil {
43+
t.Fatalf("Failed to create protocol: %v", err)
44+
}
45+
46+
if p.subscribeOption != subscribeOpt {
47+
t.Error("Subscribe option was not set correctly")
48+
}
49+
50+
if p.subscribeOption.Source != "test-source" {
51+
t.Errorf("Expected source 'test-source', got '%s'", p.subscribeOption.Source)
52+
}
53+
54+
// Test nil subscribe option
55+
_, err = NewProtocol(conn, WithSubscribeOption(nil))
56+
if err == nil {
57+
t.Error("Expected error for nil subscribe option")
58+
}
59+
}
60+
61+
func TestWithHealthCheck(t *testing.T) {
62+
lis := bufconn.Listen(1024)
63+
defer lis.Close()
64+
65+
s := grpc.NewServer()
66+
defer s.Stop()
67+
68+
conn, err := grpc.NewClient("passthrough:///bufnet",
69+
grpc.WithTransportCredentials(insecure.NewCredentials()),
70+
grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
71+
return lis.Dial()
72+
}))
73+
if err != nil {
74+
t.Fatalf("Failed to create connection: %v", err)
75+
}
76+
defer conn.Close()
77+
78+
// Wait for connection to be ready
79+
time.Sleep(50 * time.Millisecond)
80+
81+
// Test WithHealthCheck
82+
errorChan := make(chan error, 1)
83+
interval := 3 * time.Second
84+
85+
p, err := NewProtocol(conn, WithHealthCheck(interval, errorChan))
86+
if err != nil {
87+
t.Fatalf("Failed to create protocol: %v", err)
88+
}
89+
90+
if p.reconnectErrorChan != errorChan {
91+
t.Error("Health check error channel was not set correctly")
92+
}
93+
94+
if p.checkInterval != interval {
95+
t.Errorf("Expected health check interval %v, got %v", interval, p.checkInterval)
96+
}
97+
98+
// Test nil error channel for health check
99+
_, err = NewProtocol(conn, WithHealthCheck(interval, nil))
100+
if err == nil {
101+
t.Error("Expected error for nil health check error channel")
102+
}
103+
}
104+
105+
func TestMultipleOptions(t *testing.T) {
106+
lis := bufconn.Listen(1024)
107+
defer lis.Close()
108+
109+
s := grpc.NewServer()
110+
defer s.Stop()
111+
112+
conn, err := grpc.NewClient("passthrough:///bufnet",
113+
grpc.WithTransportCredentials(insecure.NewCredentials()),
114+
grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
115+
return lis.Dial()
116+
}))
117+
if err != nil {
118+
t.Fatalf("Failed to create connection: %v", err)
119+
}
120+
defer conn.Close()
121+
122+
// Wait for connection to be ready
123+
time.Sleep(50 * time.Millisecond)
124+
125+
// Test applying multiple options
126+
subscribeOpt := &SubscribeOption{
127+
Source: "test-source",
128+
DataType: "test-type",
129+
}
130+
errorChan := make(chan error, 1)
131+
interval := 2 * time.Second
132+
133+
p, err := NewProtocol(conn,
134+
WithSubscribeOption(subscribeOpt),
135+
WithHealthCheck(interval, errorChan),
136+
)
137+
if err != nil {
138+
t.Fatalf("Failed to create protocol: %v", err)
139+
}
140+
141+
// Verify both options were applied
142+
if p.subscribeOption != subscribeOpt {
143+
t.Error("Subscribe option was not set correctly")
144+
}
145+
146+
if p.reconnectErrorChan != errorChan {
147+
t.Error("Health check error channel was not set correctly")
148+
}
149+
150+
if p.checkInterval != interval {
151+
t.Errorf("Expected health check interval %v, got %v", interval, p.checkInterval)
152+
}
153+
}

‎pkg/cloudevents/generic/options/grpc/protocol/protocol.go‎

Lines changed: 25 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ type Protocol struct {
3333

3434
closeChan chan struct{}
3535

36-
// errorChan is to send an error messsage to restart the connection
36+
// errorChan is to send an error message to restart the connection
3737
reconnectErrorChan chan error
3838
checkInterval time.Duration
3939
}
@@ -184,24 +184,33 @@ func (p *Protocol) healthCheck(ctx context.Context) {
184184
healthClient := healthpb.NewHealthClient(p.clientConn)
185185
stream, err := healthClient.Watch(watchCtx, &healthpb.HealthCheckRequest{Service: ""})
186186
if err != nil {
187+
if s, ok := status.FromError(err); ok && s.Code() == codes.Unimplemented {
188+
logger.Warnf("grpc-health Watch not implemented on server; skipping health check")
189+
return
190+
}
191+
logger.Errorf("failed to watch health check, %v", err)
187192
select {
188193
case p.reconnectErrorChan <- err:
189194
default:
190195
}
191196
return
192197
}
193198

194-
last := time.Now()
195199
recvErr := make(chan error, 1)
200+
healthReceived := make(chan struct{}, 1)
196201
go func() {
197202
for {
198203
resp, err := stream.Recv()
199204
if err != nil {
205+
// Forward all errors; main loop decides whether to propagate.
200206
recvErr <- err
201207
return
202208
}
203-
last = time.Now()
204209
logger.Infof("Received server health status %s", resp.Status)
210+
select {
211+
case healthReceived <- struct{}{}:
212+
default:
213+
}
205214
}
206215
}()
207216

@@ -213,19 +222,26 @@ func (p *Protocol) healthCheck(ctx context.Context) {
213222
case <-ctx.Done():
214223
return
215224
case err := <-recvErr:
225+
if s, ok := status.FromError(err); ok && s.Code() == codes.Unimplemented {
226+
logger.Warnf("grpc-health not implemented on server; skipping health check")
227+
return
228+
}
216229
select {
217230
case p.reconnectErrorChan <- err:
218231
default:
219232
}
220233
return
234+
case <-healthReceived:
235+
// Reset the ticker when we receive a health update
236+
ticker.Stop()
237+
ticker = time.NewTicker(p.checkInterval)
221238
case <-ticker.C:
222-
if time.Since(last) > p.checkInterval {
223-
select {
224-
case p.reconnectErrorChan <- fmt.Errorf("timeout waiting for health check"):
225-
default:
226-
}
227-
return
239+
// Timeout waiting for health check
240+
select {
241+
case p.reconnectErrorChan <- fmt.Errorf("timeout waiting for health check"):
242+
default:
228243
}
244+
return
229245
}
230246
}
231247
}

0 commit comments

Comments
 (0)