Added Support for K_CA_CERTS in the heartbeats (#6920)

Signed-off-by: Vishal Choudhary <sendtovishalchoudhary@gmail.com>
Fixes #6917

<!-- Please include the 'why' behind your changes if no issue exists -->

The cmd/heartbeats component now supports Eventing TLS by using the
K_CA_CERTS environment variable in client.

## Proposed Changes

<!-- Please categorize your changes:
- 🎁 Add new feature
- 🐛 Fix bug
- 🧹 Update or clean up current behavior
- 🗑️ Remove feature or internal logic
-->

- 🧹 Update or clean up current behavior
- Added K_CA_CERTS env variable support
- Added support for eventingtls in heartbeats

### Pre-review Checklist

<!-- If these boxes are not checked, you will be asked to complete these
requirements or explain why they do not apply to your PR. -->

- [x] **At least 80% unit test coverage**
- [ ] **E2E tests** for any new behavior
- [ ] **Docs PR** for any user-facing impact
- [ ] **Spec PR** for any new API feature
- [ ] **Conformance test** for any change to the spec

**Release Note**

<!--
📄 If this change has user-visible impact, write a release
note in the block
below. Include the string "action required" if additional action is
required of
users switching to the new release, for example in case of a breaking
change.

Write as if you are speaking to users, not other Knative contributors.
If this
change has no user-visible impact, no release note is needed.
-->

```release-note
The ApiServerSource controller now sets the K_CA_CERTS environment variable when creating the adapter and the sink has CACerts defined.
```


**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->

---------

Signed-off-by: Vishal Choudhary <sendtovishalchoudhary@gmail.com>
Co-authored-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

cmd/heartbeats/main.go

@@ -28,11 +28,14 @@ import (
 	"syscall"
 	"time"
 
+	"go.opencensus.io/plugin/ochttp"
 	"go.uber.org/zap"
+	"knative.dev/eventing/pkg/eventingtls"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/tracing"
 	"knative.dev/pkg/tracing/config"
+	"knative.dev/pkg/tracing/propagation/tracecontextb3"
 
 	"github.com/cloudevents/sdk-go/observability/opencensus/v2/client"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -50,6 +53,7 @@ var (
 	eventSource string
 	eventType   string
 	sink        string
+	cacerts     string
 	label       string
 	periodStr   string
 	msg         string
@@ -59,6 +63,7 @@ func init() {
 	flag.StringVar(&eventSource, "eventSource", "", "the event-source (CloudEvents)")
 	flag.StringVar(&eventType, "eventType", "dev.knative.eventing.samples.heartbeat", "the event-type (CloudEvents)")
 	flag.StringVar(&sink, "sink", "", "the host url to heartbeat to")
+	flag.StringVar(&cacerts, "cacerts", "", "the ca cert for the host url to heartbeat to")
 	flag.StringVar(&label, "label", "", "a special label")
 	flag.StringVar(&periodStr, "period", "5s", "the duration between heartbeats. Supported formats: Go (https://pkg.go.dev/time#ParseDuration), integers (interpreted as seconds)")
 	flag.StringVar(&msg, "msg", "", "message content in data.msg")
@@ -68,6 +73,9 @@ type envConfig struct {
 	// Sink URL where to send heartbeat cloudevents
 	Sink string `envconfig:"K_SINK"`
 
+	// CACert is the certificate for enabling HTTPS in Sink URL
+	CACerts string `envconfig:"K_CA_CERTS"`
+
 	// CEOverrides are the CloudEvents overrides to be applied to the outbound event.
 	CEOverrides string `envconfig:"K_CE_OVERRIDES"`
 
@@ -102,6 +110,10 @@ func main() {
 		sink = env.Sink
 	}
 
+	if env.CACerts != "" {
+		cacerts = env.CACerts
+	}
+
 	var ceOverrides *duckv1.CloudEventOverrides
 	if len(env.CEOverrides) > 0 {
 		overrides := duckv1.CloudEventOverrides{}
@@ -122,7 +134,28 @@ func main() {
 		log.Fatalf("Failed to initialize tracing: %v", err)
 	}
 	defer tracer.Shutdown(ctx)
-	c, err := client.NewClientHTTP([]cehttp.Option{cloudevents.WithTarget(sink)}, nil)
+
+	opts := make([]cehttp.Option, 0, 1)
+	opts = append(opts, cloudevents.WithTarget(sink))
+
+	if eventingtls.IsHttpsSink(sink) {
+		clientConfig := eventingtls.NewDefaultClientConfig()
+		clientConfig.CACerts = &cacerts
+
+		httpTransport := http.DefaultTransport.(*http.Transport).Clone()
+		httpTransport.TLSClientConfig, err = eventingtls.GetTLSClientConfig(clientConfig)
+		if err != nil {
+			log.Fatalf("Failed to get TLS Client Config: %v", err)
+		}
+
+		transport := &ochttp.Transport{
+			Base:        httpTransport,
+			Propagation: tracecontextb3.TraceContextEgress,
+		}
+		opts = append(opts, cehttp.WithRoundTripper(transport))
+	}
+
+	c, err := client.NewClientHTTP(opts, nil)
 	if err != nil {
 		log.Fatalf("failed to create client: %s", err.Error())
 	}