| 85 | } |
| 86 | |
| 87 | func (w *Worker) sendHttpCloudEvent(logReq LogRequest) error { |
| 88 | t, err := cloudevents.NewHTTP( |
| 89 | cloudevents.WithTarget(logReq.Url.String()), |
| 90 | ) |
| 91 | if err != nil { |
| 92 | return fmt.Errorf("while creating http transport: %w", err) |
| 93 | } |
| 94 | |
| 95 | if logReq.Url.Scheme == "https" { |
| 96 | caCertFilePath := filepath.Join(constants.LoggerCaCertMountPath, logReq.CertName) |
| 97 | caCertFile, err := os.ReadFile(filepath.Clean(caCertFilePath)) |
| 98 | // Do not fail if certificates not found, for backwards compatibility |
| 99 | if err == nil { |
| 100 | clientCertPool := x509.NewCertPool() |
| 101 | if !clientCertPool.AppendCertsFromPEM(caCertFile) { |
| 102 | return errors.New("while parsing CA certificate") |
| 103 | } |
| 104 | |
| 105 | tlsTransport := &http.Transport{ |
| 106 | TLSClientConfig: &tls.Config{ |
| 107 | RootCAs: clientCertPool, |
| 108 | MinVersion: tls.VersionTLS12, |
| 109 | InsecureSkipVerify: logReq.TlsSkipVerify, // #nosec G402 |
| 110 | }, |
| 111 | } |
| 112 | t.Client.Transport = tlsTransport |
| 113 | } else { |
| 114 | w.Log.Warnf("using https endpoint but could not find CA cert file %s", caCertFilePath) |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | c, err := cloudevents.NewClient(t) |
| 119 | if err != nil { |
| 120 | return fmt.Errorf("while creating new cloudevents client: %w", err) |
| 121 | } |
| 122 | event := cloudevents.NewEvent(cloudevents.VersionV1) |
| 123 | event.SetID(logReq.Id) |
| 124 | event.SetType(logReq.ReqType) |
| 125 | event.SetTime(logReq.OccurrenceTime) |
| 126 | event.SetExtension(RecordedTimeAttr, time.Now()) |
| 127 | |
| 128 | event.SetExtension(InferenceServiceAttr, logReq.InferenceService) |
| 129 | event.SetExtension(NamespaceAttr, logReq.Namespace) |
| 130 | event.SetExtension(ComponentAttr, logReq.Component) |
| 131 | event.SetExtension(EndpointAttr, logReq.Endpoint) |
| 132 | |
| 133 | encodedMetadata, err := json.Marshal(logReq.Metadata) |
| 134 | if err != nil { |
| 135 | return fmt.Errorf("could not encode metadata as json: %w", err) |
| 136 | } |
| 137 | event.SetExtension(MetadataAttr, string(encodedMetadata)) |
| 138 | |
| 139 | if len(logReq.Annotations) > 0 { |
| 140 | bits, err := json.Marshal(logReq.Annotations) |
| 141 | if err != nil { |
| 142 | w.Log.Errorf("failed to marshal annotations: %w", err) |
| 143 | } else { |
| 144 | event.SetExtension(AnnotationAttr, string(bits)) |