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

Commit

Permalink
[0.17] Update eventing to latest (#1657)
Browse files Browse the repository at this point in the history
Signed-off-by: Pierangelo Di Pilato <[email protected]>
  • Loading branch information
pierDipi authored Nov 6, 2020
1 parent 090e2b3 commit cfad5be
Show file tree
Hide file tree
Showing 35 changed files with 1,190 additions and 913 deletions.
2 changes: 1 addition & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ require (
k8s.io/apimachinery v0.19.0
k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
knative.dev/eventing v0.17.5-0.20200925065343-049b8e743bd4
knative.dev/eventing v0.17.9-0.20201105153307-2fb113c42ff4
knative.dev/pkg v0.0.0-20200824160247-5343c1d19369
knative.dev/serving v0.17.1
knative.dev/test-infra v0.0.0-20200828171708-f68cb78c80a9
Expand Down
4 changes: 2 additions & 2 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -2028,8 +2028,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-20200811171106-48c335fed9c8/go.mod h1:XonDcFC2DLSWP71f2y7oYnXUko5d5HsJRnZtkp0wY7g=
knative.dev/eventing v0.17.5-0.20200925065343-049b8e743bd4 h1:Wygx5VC4nZDs9p1Om8EYWqHMSNDZ6gWhio09hj0KVMw=
knative.dev/eventing v0.17.5-0.20200925065343-049b8e743bd4/go.mod h1:9NwCSwLnMCKmgz3YQBNax18mSgBjud8CvfsUUVOZ1sA=
knative.dev/eventing v0.17.9-0.20201105153307-2fb113c42ff4 h1:g6ud+UJbnjht9uciWVf29aUFAI3IKn2PfyTQpdXkD3Y=
knative.dev/eventing v0.17.9-0.20201105153307-2fb113c42ff4/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-20200812200006-4d518e76538a h1:E1rnQR9IZvDcEAgoOXMW9LWqevaYFVTlMS2ndgoAO6Y=
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,7 @@ func (ts *TriggerStatus) MarkDependencyNotConfigured() {
"DependencyNotConfigured", "Dependency has not yet been reconciled.")
}

func (ts *TriggerStatus) PropagateDependencyStatus(ks *duckv1.KResource) {
func (ts *TriggerStatus) PropagateDependencyStatus(ks *duckv1.Source) {
kc := ks.Status.GetCondition(apis.ConditionReady)
if kc == nil {
ts.MarkDependencyNotConfigured()
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 ""
}
Original file line number Diff line number Diff line change
Expand Up @@ -104,13 +104,15 @@ func BrokerChannelFlowWithTransformation(t *testing.T,
}

// create the transformation service for trigger1
transformationPod := resources.EventTransformationPod(
recordevents.DeployEventRecordOrFail(
client,
transformationPodName,
transformedEventType,
transformedEventSource,
[]byte(transformedBody),
recordevents.ReplyWithTransformedEvent(
transformedEventType,
transformedEventSource,
transformedBody,
),
)
client.CreatePodOrFail(transformationPod, testlib.WithService(transformationPodName))

// create trigger1 to receive the original event, and do event transformation
if triggerVersion == "v1" {
Expand Down
Loading

0 comments on commit cfad5be

Please sign in to comment.