You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jr...@apache.org on 2023/11/03 18:34:05 UTC
(beam) branch master updated: Add 10/1s refresher (#29295)
This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 42bd0d08950 Add 10/1s refresher (#29295)
42bd0d08950 is described below
commit 42bd0d089500ef57ac2c3a1aaa541bbf2924c8b4
Author: Damon <da...@users.noreply.github.com>
AuthorDate: Fri Nov 3 11:33:57 2023 -0700
Add 10/1s refresher (#29295)
---
.../overlays/echo-10-per-1s-quota/configmap.yaml | 30 ++++++++++++
.../overlays/echo-10-per-1s-quota/deployment.yaml | 27 +++++++++++
.../echo-10-per-1s-quota/kustomization.yaml | 34 +++++++++++++
.../src/main/go/test/integration/echo/echo_test.go | 56 +++++++++++++++++-----
.../mock-apis/src/main/go/test/integration/vars.go | 12 ++---
5 files changed, 140 insertions(+), 19 deletions(-)
diff --git a/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/configmap.yaml b/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/configmap.yaml
new file mode 100644
index 00000000000..d78c862c2af
--- /dev/null
+++ b/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/configmap.yaml
@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Configures patch for ../base/configmap.yaml
+# See https://kubectl.docs.kubernetes.io/references/kustomize/kustomization/patches/
+
+- op: replace
+ path: /metadata/labels/quota-id
+ value: echo-10-per-1s-quota
+- op: replace
+ path: /data/QUOTA_ID
+ value: echo-10-per-1s-quota
+- op: replace
+ path: /data/QUOTA_SIZE
+ value: "10"
+- op: replace
+ path: /data/QUOTA_REFRESH_INTERVAL
+ value: 1s
diff --git a/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/deployment.yaml b/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/deployment.yaml
new file mode 100644
index 00000000000..37af48ec97d
--- /dev/null
+++ b/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/deployment.yaml
@@ -0,0 +1,27 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Configures patch for ../base/deployment.yaml
+# See https://kubectl.docs.kubernetes.io/references/kustomize/kustomization/patches/
+
+- op: replace
+ path: /metadata/labels/quota-id
+ value: echo-10-per-1s-quota
+- op: replace
+ path: /spec/selector/matchLabels/quota-id
+ value: echo-10-per-1s-quota
+- op: replace
+ path: /spec/template/metadata/labels/quota-id
+ value: echo-10-per-1s-quota
diff --git a/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/kustomization.yaml b/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/kustomization.yaml
new file mode 100644
index 00000000000..d5069835443
--- /dev/null
+++ b/.test-infra/mock-apis/infrastructure/kubernetes/refresher/overlays/echo-10-per-1s-quota/kustomization.yaml
@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Configures the overlay for .test-infra/mock-apis/infrastructure/kubernetes/refresher/base
+# Using the Quota Id:
+# echo-10-per-1s-quota
+
+resources:
+- ../../base
+
+nameSuffix: -echo-10-per-1s-quota
+
+patches:
+- path: configmap.yaml
+ target:
+ kind: ConfigMap
+ name: refresher
+
+- path: deployment.yaml
+ target:
+ kind: Deployment
+ name: refresher
diff --git a/.test-infra/mock-apis/src/main/go/test/integration/echo/echo_test.go b/.test-infra/mock-apis/src/main/go/test/integration/echo/echo_test.go
index 73ad597c7d3..102d8508d48 100644
--- a/.test-infra/mock-apis/src/main/go/test/integration/echo/echo_test.go
+++ b/.test-infra/mock-apis/src/main/go/test/integration/echo/echo_test.go
@@ -24,6 +24,7 @@ import (
"fmt"
"net/http"
"regexp"
+ "sync"
"testing"
"time"
@@ -45,6 +46,8 @@ const (
shouldExceedQuotaId = "echo-should-exceed-quota"
shouldNeverExceedQuotaId = "echo-should-never-exceed-quota"
shouldNotExistId = "should-not-exist"
+ refresh10Per1s = "echo-10-per-1s-quota"
+ defaultNumCalls = 3
)
var (
@@ -60,11 +63,12 @@ func TestEcho(t *testing.T) {
payload := []byte("payload")
for _, tt := range []struct {
- tag string
- quotaId string
- client echov1.EchoServiceClient
- want *echov1.EchoResponse
- wantErr error
+ tag string
+ quotaId string
+ client echov1.EchoServiceClient
+ want *echov1.EchoResponse
+ numCalls int
+ wantErr error
}{
{
tag: "http",
@@ -108,11 +112,25 @@ func TestEcho(t *testing.T) {
Payload: payload,
},
},
+ {
+ numCalls: 20,
+ tag: "grpc",
+ quotaId: refresh10Per1s,
+ client: withGrpc(t),
+ wantErr: status.Error(codes.ResourceExhausted, "error: resource exhausted for: echo-10-per-1s-quota"),
+ },
} {
t.Run(fmt.Sprintf("%s/%s", tt.quotaId, tt.tag), func(t *testing.T) {
ctx, cancel := withTimeout()
defer cancel()
+ if tt.numCalls == 0 {
+ tt.numCalls = defaultNumCalls
+ }
+
+ wg := sync.WaitGroup{}
+ wg.Add(tt.numCalls)
+
req := &echov1.EchoRequest{
Id: tt.quotaId,
Payload: payload,
@@ -121,16 +139,21 @@ func TestEcho(t *testing.T) {
var resps []*echov1.EchoResponse
var errs []error
- for i := 0; i < 3; i++ {
- resp, err := tt.client.Echo(ctx, req)
- if err != nil {
- errs = append(errs, err)
- }
- if resp != nil {
- resps = append(resps, resp)
- }
+ for i := 0; i < tt.numCalls; i++ {
+ go func() {
+ resp, err := tt.client.Echo(ctx, req)
+ if err != nil {
+ errs = append(errs, err)
+ }
+ if resp != nil {
+ resps = append(resps, resp)
+ }
+ wg.Done()
+ }()
}
+ wg.Wait()
+
if tt.wantErr != nil && len(errs) == 0 {
t.Errorf("Echo(%+v) err = nil, wantErr = %v", req, tt.wantErr)
return
@@ -163,6 +186,10 @@ func withGrpc(t *testing.T) echov1.EchoServiceClient {
ctx, cancel := withTimeout()
defer cancel()
+ if *integration.GRPCServiceEndpoint == "" {
+ t.Fatalf("missing flag: -%s", integration.GrpcServiceEndpointFlag)
+ }
+
conn, err := grpc.DialContext(ctx, *integration.GRPCServiceEndpoint, grpcOpts...)
if err != nil {
t.Fatalf("DialContext(%s) err %v", *integration.GRPCServiceEndpoint, err)
@@ -205,6 +232,9 @@ func (h *httpCaller) Echo(ctx context.Context, in *echov1.EchoRequest, _ ...grpc
}
func withHttp(t *testing.T) echov1.EchoServiceClient {
+ if *integration.HTTPServiceEndpoint == "" {
+ t.Fatalf("missing flag: -%s", integration.HttpServiceEndpointFlag)
+ }
p := regexp.MustCompile(`^http://`)
rawUrl := fmt.Sprint(*integration.HTTPServiceEndpoint, echo.PathAlias)
if !p.MatchString(rawUrl) {
diff --git a/.test-infra/mock-apis/src/main/go/test/integration/vars.go b/.test-infra/mock-apis/src/main/go/test/integration/vars.go
index 92ba445fdfa..a4bd9bb60e2 100644
--- a/.test-infra/mock-apis/src/main/go/test/integration/vars.go
+++ b/.test-infra/mock-apis/src/main/go/test/integration/vars.go
@@ -21,8 +21,8 @@ import (
)
const (
- grpcServiceEndpointFlag = "grpc_service_endpoint"
- httpServiceEndpointFlag = "http_service_endpoint"
+ GrpcServiceEndpointFlag = "grpc_service_endpoint"
+ HttpServiceEndpointFlag = "http_service_endpoint"
moreInfoUrl = "https://github.com/apache/beam/tree/master/.test-infra/mock-apis#writing-integration-tests"
)
@@ -31,8 +31,8 @@ var (
moreInfo = fmt.Sprintf("See %s for more information on how to get the relevant value for your test.", moreInfoUrl)
requiredFlags = []string{
- grpcServiceEndpointFlag,
- httpServiceEndpointFlag,
+ GrpcServiceEndpointFlag,
+ HttpServiceEndpointFlag,
}
)
@@ -40,10 +40,10 @@ var (
// go test ./src/main/go/test/integration/...
var (
// GRPCServiceEndpoint is the address of the deployed service.
- GRPCServiceEndpoint = flag.String(grpcServiceEndpointFlag, "",
+ GRPCServiceEndpoint = flag.String(GrpcServiceEndpointFlag, "",
"The endpoint to target gRPC calls to a service. "+moreInfo)
// HTTPServiceEndpoint is the address of the deployed service.
- HTTPServiceEndpoint = flag.String(httpServiceEndpointFlag, "",
+ HTTPServiceEndpoint = flag.String(HttpServiceEndpointFlag, "",
"The endpoint to target HTTP calls to a service. "+moreInfo)
)