Skip to content
This repository has been archived by the owner on Jun 19, 2022. It is now read-only.

Commit

Permalink
update deps to pick up eventing #4475 fix (#1889)
Browse files Browse the repository at this point in the history
  • Loading branch information
capri-xiyue authored Nov 6, 2020
1 parent 90d4535 commit 7ed189d
Show file tree
Hide file tree
Showing 35 changed files with 1,197 additions and 919 deletions.
6 changes: 3 additions & 3 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,9 @@ require (
k8s.io/api v0.18.7-rc.0
k8s.io/apimachinery v0.19.1
k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
knative.dev/eventing v0.17.7
knative.dev/pkg v0.0.0-20200831162708-14fb2347fb77
knative.dev/serving v0.17.4-0.20201013135631-4ed85f328cd3
knative.dev/eventing v0.17.9-0.20201106045707-9ce198d8d46f
knative.dev/pkg v0.0.0-20201026133041-d87c75f8b67d
knative.dev/serving v0.17.4
sigs.k8s.io/yaml v1.2.0
)

Expand Down
10 changes: 6 additions & 4 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -1972,8 +1972,8 @@ k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
knative.dev/caching v0.0.0-20190719140829-2032732871ff/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
knative.dev/caching v0.0.0-20200116200605-67bca2c83dfa/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
knative.dev/caching v0.0.0-20200831163411-f6154e7455e2/go.mod h1:jG64HWkBQmFlY8dk1CH884oHtaT/savnqxcZC8vBnvk=
knative.dev/eventing v0.17.7 h1:1zjPta8FDZ8Z0VosaeEWqAmWd199nd+4JrHfkO0XviE=
knative.dev/eventing v0.17.7/go.mod h1:9NwCSwLnMCKmgz3YQBNax18mSgBjud8CvfsUUVOZ1sA=
knative.dev/eventing v0.17.9-0.20201106045707-9ce198d8d46f h1:wJIv+B9YmMGYSW1IRnYxSpi0TSvGQr8zN/+ngFKwD0c=
knative.dev/eventing v0.17.9-0.20201106045707-9ce198d8d46f/go.mod h1:9NwCSwLnMCKmgz3YQBNax18mSgBjud8CvfsUUVOZ1sA=
knative.dev/eventing-contrib v0.6.1-0.20190723221543-5ce18048c08b/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
knative.dev/eventing-contrib v0.11.2/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
knative.dev/networking v0.0.0-20200831172815-5f2e0ad6215f h1:Ru8WsB1JM2O8YOLFq6N6r4p0RAb2Foo+1nGVlVnDvqY=
Expand All @@ -1992,8 +1992,10 @@ knative.dev/pkg v0.0.0-20200824160247-5343c1d19369 h1:1hf89VqyCXYwYUueJ+zMrbvbm9
knative.dev/pkg v0.0.0-20200824160247-5343c1d19369/go.mod h1:7Ezp2dJ/stJ4tyrPJ5Loetb+iiwiXNAt+YLrtzE/y4E=
knative.dev/pkg v0.0.0-20200831162708-14fb2347fb77 h1:GVzuZIcWF2FqflS8gEwSYWO3bF+0qvx47dzMlBxnJIo=
knative.dev/pkg v0.0.0-20200831162708-14fb2347fb77/go.mod h1:xLMZC5ct0Sx3QQDWHMa1WZggjR6lpJt/S7mXEED6ONo=
knative.dev/serving v0.17.4-0.20201013135631-4ed85f328cd3 h1:hrIZGtwSsetA73ZDsvtKhhAxsM0CoKCxgBNX2W1vfus=
knative.dev/serving v0.17.4-0.20201013135631-4ed85f328cd3/go.mod h1:AT6FFxMBYp6kE30wfA/OspfCBTizrjIpaAcRLo2JtwA=
knative.dev/pkg v0.0.0-20201026133041-d87c75f8b67d h1:dCpKVhcYFJS086dy3dK0g5vjeYYyvUF9OTNr7i4qsP4=
knative.dev/pkg v0.0.0-20201026133041-d87c75f8b67d/go.mod h1:xLMZC5ct0Sx3QQDWHMa1WZggjR6lpJt/S7mXEED6ONo=
knative.dev/serving v0.17.4 h1:Ym90IC+08YBq9cbX0lkGzn+hLr0E3ZWWYkxvhHxj6WE=
knative.dev/serving v0.17.4/go.mod h1:AT6FFxMBYp6kE30wfA/OspfCBTizrjIpaAcRLo2JtwA=
knative.dev/test-infra v0.0.0-20200407185800-1b88cb3b45a5/go.mod h1:xcdUkMJrLlBswIZqL5zCuBFOC22WIPMQoVX1L35i0vQ=
knative.dev/test-infra v0.0.0-20200505052144-5ea2f705bb55/go.mod h1:WqF1Azka+FxPZ20keR2zCNtiQA1MP9ZB4BH4HuI+SIU=
knative.dev/test-infra v0.0.0-20200513011557-d03429a76034/go.mod h1:aMif0KXL4g19YCYwsy4Ocjjz5xgPlseYV+B95Oo4JGE=
Expand Down
106 changes: 106 additions & 0 deletions vendor/knative.dev/eventing/pkg/kncloudevents/http_client.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
/*
Copyright 2020 The Knative Authors
Licensed 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.
*/

package kncloudevents

import (
nethttp "net/http"
"sync"
"time"

"go.opencensus.io/plugin/ochttp"
"knative.dev/pkg/tracing/propagation/tracecontextb3"
)

const (
defaultRetryWaitMin = 1 * time.Second
defaultRetryWaitMax = 30 * time.Second
)

type holder struct {
clientMutex sync.Mutex
connectionArgs *ConnectionArgs
client **nethttp.Client
}

var clientHolder = holder{}

// The used HTTP client is a singleton, so the same http client is reused across all the application.
// If connection args is modified, client is cleaned and a new one is created.
func getClient() *nethttp.Client {
clientHolder.clientMutex.Lock()
defer clientHolder.clientMutex.Unlock()

if clientHolder.client == nil {
// Add connection options to the default transport.
var base = nethttp.DefaultTransport.(*nethttp.Transport).Clone()
clientHolder.connectionArgs.configureTransport(base)
c := &nethttp.Client{
// Add output tracing.
Transport: &ochttp.Transport{
Base: base,
Propagation: tracecontextb3.TraceContextEgress,
},
}
clientHolder.client = &c
}

return *clientHolder.client
}

// ConfigureConnectionArgs configures the new connection args.
// The existing client won't be affected, but a new one will be created.
// Use sparingly, because it might lead to creating a lot of clients, none of them sharing their connection pool!
func ConfigureConnectionArgs(ca *ConnectionArgs) {
clientHolder.clientMutex.Lock()
defer clientHolder.clientMutex.Unlock()

// Check if same config
if clientHolder.connectionArgs != nil &&
ca != nil &&
ca.MaxIdleConns == clientHolder.connectionArgs.MaxIdleConns &&
ca.MaxIdleConnsPerHost == clientHolder.connectionArgs.MaxIdleConnsPerHost {
return
}

if clientHolder.client != nil {
// Let's try to clean up a bit the existing client
// Note: this won't remove it nor close it
(*clientHolder.client).CloseIdleConnections()

// Setting client to nil
clientHolder.client = nil
}

clientHolder.connectionArgs = ca
}

// ConnectionArgs allow to configure connection parameters to the underlying
// HTTP Client transport.
type ConnectionArgs struct {
// MaxIdleConns refers to the max idle connections, as in net/http/transport.
MaxIdleConns int
// MaxIdleConnsPerHost refers to the max idle connections per host, as in net/http/transport.
MaxIdleConnsPerHost int
}

func (ca *ConnectionArgs) configureTransport(transport *nethttp.Transport) {
if ca == nil {
return
}
transport.MaxIdleConns = ca.MaxIdleConns
transport.MaxIdleConnsPerHost = ca.MaxIdleConnsPerHost
}
52 changes: 12 additions & 40 deletions vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,17 +25,10 @@ import (

"github.com/hashicorp/go-retryablehttp"
"github.com/rickb777/date/period"
"go.opencensus.io/plugin/ochttp"
"knative.dev/pkg/tracing/propagation/tracecontextb3"

duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
)

const (
defaultRetryWaitMin = 1 * time.Second
defaultRetryWaitMax = 30 * time.Second
)

var noRetries = RetryConfig{
RetryMax: 0,
CheckRetry: func(ctx context.Context, resp *nethttp.Response, err error) (bool, error) {
Expand All @@ -46,41 +39,15 @@ var noRetries = RetryConfig{
},
}

// ConnectionArgs allow to configure connection parameters to the underlying
// HTTP Client transport.
type ConnectionArgs struct {
// MaxIdleConns refers to the max idle connections, as in net/http/transport.
MaxIdleConns int
// MaxIdleConnsPerHost refers to the max idle connections per host, as in net/http/transport.
MaxIdleConnsPerHost int
}

func (ca *ConnectionArgs) ConfigureTransport(transport *nethttp.Transport) {
if ca == nil {
return
}
transport.MaxIdleConns = ca.MaxIdleConns
transport.MaxIdleConnsPerHost = ca.MaxIdleConnsPerHost
}

type HttpMessageSender struct {
Client *nethttp.Client
Target string
}

func NewHttpMessageSender(connectionArgs *ConnectionArgs, target string) (*HttpMessageSender, error) {
// Add connection options to the default transport.
var base = nethttp.DefaultTransport.(*nethttp.Transport).Clone()
connectionArgs.ConfigureTransport(base)
// Add output tracing.
client := &nethttp.Client{
Transport: &ochttp.Transport{
Base: base,
Propagation: tracecontextb3.TraceContextEgress,
},
}

return &HttpMessageSender{Client: client, Target: target}, nil
// Deprecated: Don't use this anymore, now it has the same effect of NewHTTPMessageSenderWithTarget
// If you need to modify the connection args, use ConfigureConnectionArgs sparingly.
func NewHttpMessageSender(ca *ConnectionArgs, target string) (*HttpMessageSender, error) {
return &HttpMessageSender{Client: getClient(), Target: target}, nil
}

func (s *HttpMessageSender) NewCloudEventRequest(ctx context.Context) (*nethttp.Request, error) {
Expand Down Expand Up @@ -139,7 +106,12 @@ func (s *HttpMessageSender) SendWithRetries(req *nethttp.Request, config *RetryC
},
}

return retryableClient.Do(&retryablehttp.Request{Request: req})
retryableReq, err := retryablehttp.FromRequest(req)
if err != nil {
return nil, err
}

return retryableClient.Do(retryableReq)
}

func NoRetries() RetryConfig {
Expand Down Expand Up @@ -179,6 +151,6 @@ func RetryConfigFromDeliverySpec(spec duckv1.DeliverySpec) (RetryConfig, error)
return retryConfig, nil
}

func checkRetry(_ context.Context, resp *nethttp.Response, _ error) (bool, error) {
return resp != nil && resp.StatusCode >= 300, nil
func checkRetry(_ context.Context, resp *nethttp.Response, err error) (bool, error) {
return !(resp != nil && resp.StatusCode < 300), err
}
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import (
"knative.dev/pkg/reconciler"

eventingv1beta1 "knative.dev/eventing/pkg/apis/eventing/v1beta1"
"knative.dev/eventing/test/lib/recordevents"

testlib "knative.dev/eventing/test/lib"
"knative.dev/eventing/test/lib/duck"
Expand Down Expand Up @@ -81,8 +82,7 @@ func triggerV1Beta1BeforeBrokerHelper(triggerName string, client *testlib.Client
const etLogger = "logger"
const loggerPodName = "logger-pod"

logPod := resources.EventRecordPod(loggerPodName)
client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
_ = recordevents.DeployEventRecordOrFail(client, loggerPodName)
client.WaitForAllTestResourcesReadyOrFail() // Can't do this for the trigger because it's not 'ready' yet
client.CreateTriggerOrFailV1Beta1(triggerName,
resources.WithAttributesTriggerFilterV1Beta1(eventingv1beta1.TriggerAnyFilter, etLogger, map[string]interface{}{}),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -316,13 +316,15 @@ func BrokerV1Beta1ConsumerDataPlaneTestHelper(
source := "origin-for-reply"
event.SetSource(source)
msg := []byte(`{"msg":"Transformed!"}`)
transformPod := resources.EventTransformationPod(
recordevents.DeployEventRecordOrFail(
client,
"transformer-pod",
"reply-check-type",
"reply-check-source",
msg,
recordevents.ReplyWithTransformedEvent(
"reply-check-type",
"reply-check-source",
string(msg),
),
)
client.CreatePodOrFail(transformPod, testlib.WithService("transformer-pod"))
client.WaitForServiceEndpointsOrFail("transformer-pod", 1)
transformTrigger := client.CreateTriggerOrFailV1Beta1(
"transform-trigger",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import (
"knative.dev/eventing/pkg/utils"
tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
testlib "knative.dev/eventing/test/lib"
"knative.dev/eventing/test/lib/recordevents"
"knative.dev/eventing/test/lib/resources"
"knative.dev/eventing/test/lib/sender"
)
Expand Down Expand Up @@ -78,8 +79,7 @@ func setupBrokerTracing(brokerClass string) SetupTracingTestInfrastructureFunc {
)

// Create a logger (EventRecord) Pod and a K8s Service that points to it.
logPod := resources.EventRecordPod(loggerPodName)
client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
_ = recordevents.DeployEventRecordOrFail(client, loggerPodName)

// Create a Trigger that receives events (type=bar) and sends them to the logger Pod.
loggerTrigger := client.CreateTriggerOrFailV1Beta1(
Expand All @@ -89,15 +89,17 @@ func setupBrokerTracing(brokerClass string) SetupTracingTestInfrastructureFunc {
resources.WithSubscriberServiceRefForTriggerV1Beta1(loggerPodName),
)

// Create a transformer (EventTransfrmer) Pod that replies with the same event as the input,
// Create a transformer Pod (recordevents with transform reply) that replies with the same event as the input,
// except the reply's event's type is changed to bar.
eventTransformerPod := resources.EventTransformationPod(
eventTransformerPod := recordevents.DeployEventRecordOrFail(
client,
"transformer",
etLogger,
senderName,
[]byte(eventBody),
recordevents.ReplyWithTransformedEvent(
etLogger,
senderName,
eventBody,
),
)
client.CreatePodOrFail(eventTransformerPod, testlib.WithService(eventTransformerPod.Name))

// Create a Trigger that receives events (type=foo) and sends them to the transformer Pod.
transformerTrigger := client.CreateTriggerOrFailV1Beta1(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
duckv1beta1 "knative.dev/eventing/pkg/apis/duck/v1beta1"
eventingv1beta1 "knative.dev/eventing/pkg/apis/messaging/v1beta1"
testlib "knative.dev/eventing/test/lib"
"knative.dev/eventing/test/lib/recordevents"
"knative.dev/eventing/test/lib/resources"

corev1 "k8s.io/api/core/v1"
Expand Down Expand Up @@ -56,8 +57,7 @@ func channelHasRequiredSubscriberStatus(st *testing.T, client *testlib.Client, c
client.CreateChannelOrFail(channelName, &channel)
client.WaitForResourceReadyOrFail(channelName, &channel)

pod := resources.EventRecordPod(subscriberServiceName + "-pod")
client.CreatePodOrFail(pod, testlib.WithService(subscriberServiceName))
_ = recordevents.DeployEventRecordOrFail(client, subscriberServiceName+"-pod")

subscription := client.CreateSubscriptionOrFail(
subscriberServiceName,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import (

tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
testlib "knative.dev/eventing/test/lib"
"knative.dev/eventing/test/lib/recordevents"
"knative.dev/eventing/test/lib/resources"
"knative.dev/eventing/test/lib/sender"
)
Expand Down Expand Up @@ -66,17 +67,18 @@ func setupChannelTracingWithReply(
client.CreateChannelOrFail(replyChannelName, channel)

// Create the 'sink', a LogEvents Pod and a K8s Service that points to it.
recordEventsPod := resources.EventRecordPod(recordEventsPodName)
client.CreatePodOrFail(recordEventsPod, testlib.WithService(recordEventsPodName))
recordEventsPod := recordevents.DeployEventRecordOrFail(client, recordEventsPodName)

// Create the subscriber, a Pod that mutates the event.
transformerPod := resources.EventTransformationPod(
transformerPod := recordevents.DeployEventRecordOrFail(
client,
"transformer",
"mutated",
eventSource,
nil,
recordevents.ReplyWithTransformedEvent(
"mutated",
eventSource,
"",
),
)
client.CreatePodOrFail(transformerPod, testlib.WithService(transformerPod.Name))

// Create the Subscription linking the Channel to the mutator.
client.CreateSubscriptionOrFail(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ func tracingTest(
expectedTestSpan, eventMatcher := setupInfrastructure(t, &channel, client, recordEventsPodName, true)

// Start the event info store and assert the event was received correctly
targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName)
targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName, client.Namespace)
if err != nil {
t.Fatalf("Pod tracker failed: %v", err)
}
Expand Down Expand Up @@ -105,6 +105,6 @@ func getTraceIDHeader(t *testing.T, evInfos []recordevents.EventInfo) string {
}
}
}
t.Fatalf("FAIL: No traceid in %d messages: (%s)", len(evInfos), evInfos)
t.Fatalf("FAIL: No traceid in %d messages: (%v)", len(evInfos), evInfos)
return ""
}
Loading

0 comments on commit 7ed189d

Please sign in to comment.