|
| 1 | +/*___INFO__MARK_BEGIN__*/ |
| 2 | +/************************************************************************* |
| 3 | +* Copyright 2024 HPC-Gridware GmbH |
| 4 | +* |
| 5 | +* Licensed under the Apache License, Version 2.0 (the "License"); |
| 6 | +* you may not use this file except in compliance with the License. |
| 7 | +* You may obtain a copy of the License at |
| 8 | +* |
| 9 | +* http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | +* |
| 11 | +* Unless required by applicable law or agreed to in writing, software |
| 12 | +* distributed under the License is distributed on an "AS IS" BASIS, |
| 13 | +* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 14 | +* See the License for the specific language governing permissions and |
| 15 | +* limitations under the License. |
| 16 | +* |
| 17 | +************************************************************************/ |
| 18 | +/*___INFO__MARK_END__*/ |
| 19 | + |
| 20 | +package main |
| 21 | + |
| 22 | +import ( |
| 23 | + "context" |
| 24 | + "encoding/json" |
| 25 | + "fmt" |
| 26 | + "net/http" |
| 27 | + "reflect" |
| 28 | + "time" |
| 29 | + |
| 30 | + "go.opentelemetry.io/contrib/bridges/otelslog" |
| 31 | + "go.opentelemetry.io/otel" |
| 32 | + "go.opentelemetry.io/otel/attribute" |
| 33 | + "go.opentelemetry.io/otel/exporters/stdout/stdoutlog" |
| 34 | + "go.opentelemetry.io/otel/exporters/stdout/stdoutmetric" |
| 35 | + "go.opentelemetry.io/otel/exporters/stdout/stdouttrace" |
| 36 | + "go.opentelemetry.io/otel/log/global" |
| 37 | + "go.opentelemetry.io/otel/sdk/log" |
| 38 | + "go.opentelemetry.io/otel/sdk/metric" |
| 39 | + "go.opentelemetry.io/otel/sdk/trace" |
| 40 | +) |
| 41 | + |
| 42 | +const name = "go.hpc-gridware.com/example/qconf" |
| 43 | + |
| 44 | +var ( |
| 45 | + tracer = otel.Tracer(name) |
| 46 | + meter = otel.Meter(name) |
| 47 | + logger = otelslog.NewLogger(name) |
| 48 | +) |
| 49 | + |
| 50 | +func init() { |
| 51 | + |
| 52 | +} |
| 53 | + |
| 54 | +// Usage |
| 55 | +// router := mux.NewRouter() |
| 56 | +// router.Handle("/api/v0/command", adapter.NewAdapter(qconf)).Methods("POST") |
| 57 | + |
| 58 | +// A JSON request body is expected with the following structure: |
| 59 | +// { |
| 60 | +// "method": "<method name>", |
| 61 | +// "args": [ |
| 62 | +// "arg1", |
| 63 | +// "arg2", |
| 64 | +// ... |
| 65 | +// ] |
| 66 | +// } |
| 67 | + |
| 68 | +type CommandRequest struct { |
| 69 | + MethodName string `json:"method"` |
| 70 | + Args []json.RawMessage `json:"args"` |
| 71 | +} |
| 72 | + |
| 73 | +// NewAdapter creates an http.Handler that for any Go interface. |
| 74 | +// The method name and arguments are expected in the JSON request body. |
| 75 | +// The response is the return value of the method also in JSON format. |
| 76 | +// The arguments and the return values must have a JSON serializable format. |
| 77 | +// Only 1 or 2 return values are supported. In case of an error of the |
| 78 | +// executed function an http status code 500 is returned. |
| 79 | +// |
| 80 | +// The adapter uses OpenTelemetry to trace the method calls and log the errors. |
| 81 | +func NewAdapter(instance interface{}) http.Handler { |
| 82 | + loggerProvider, err := newLoggerProvider() |
| 83 | + if err != nil { |
| 84 | + panic(err) |
| 85 | + } |
| 86 | + global.SetLoggerProvider(loggerProvider) |
| 87 | + |
| 88 | + tracerProvider, err := newTraceProvider() |
| 89 | + if err != nil { |
| 90 | + panic(err) |
| 91 | + } |
| 92 | + otel.SetTracerProvider(tracerProvider) |
| 93 | + |
| 94 | + meterProvider, err := newMeterProvider() |
| 95 | + if err != nil { |
| 96 | + panic(err) |
| 97 | + } |
| 98 | + otel.SetMeterProvider(meterProvider) |
| 99 | + |
| 100 | + return &adapter{ |
| 101 | + instance: instance, |
| 102 | + } |
| 103 | +} |
| 104 | + |
| 105 | +type adapter struct { |
| 106 | + instance interface{} |
| 107 | +} |
| 108 | + |
| 109 | +func (a *adapter) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 110 | + ctx, span := tracer.Start(r.Context(), "request") |
| 111 | + defer span.End() |
| 112 | + |
| 113 | + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) |
| 114 | + defer cancel() |
| 115 | + |
| 116 | + var req CommandRequest |
| 117 | + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { |
| 118 | + logErr := fmt.Errorf("invalid request payload: %w", err) |
| 119 | + a.fail(ctx, w, r, http.StatusBadRequest, logErr.Error(), err) |
| 120 | + return |
| 121 | + } |
| 122 | + |
| 123 | + logger.InfoContext(ctx, "request", req.MethodName) |
| 124 | + |
| 125 | + method := reflect.ValueOf(a.instance).MethodByName(req.MethodName) |
| 126 | + if !method.IsValid() { |
| 127 | + logErr := fmt.Errorf("method not found: %s", req.MethodName) |
| 128 | + a.fail(ctx, w, r, http.StatusNotFound, logErr.Error(), nil) |
| 129 | + return |
| 130 | + } |
| 131 | + |
| 132 | + methodType := method.Type() |
| 133 | + if methodType.NumIn() != len(req.Args) { |
| 134 | + logErr := fmt.Errorf("invalid number of arguments for method %s: %d but should be %d", |
| 135 | + req.MethodName, len(req.Args), methodType.NumIn()) |
| 136 | + a.fail(ctx, w, r, http.StatusBadRequest, |
| 137 | + logErr.Error(), nil) |
| 138 | + return |
| 139 | + } |
| 140 | + |
| 141 | + args := make([]reflect.Value, len(req.Args)) |
| 142 | + for i, arg := range req.Args { |
| 143 | + argType := methodType.In(i) |
| 144 | + argValue := reflect.New(argType).Interface() |
| 145 | + if err := json.Unmarshal(arg, argValue); err != nil { |
| 146 | + a.fail(ctx, w, r, http.StatusBadRequest, |
| 147 | + fmt.Sprintf("Invalid argument %d", i), err) |
| 148 | + return |
| 149 | + } |
| 150 | + args[i] = reflect.Indirect(reflect.ValueOf(argValue)) |
| 151 | + } |
| 152 | + |
| 153 | + qconfCallValueAttr := attribute.String("qconf.command", req.MethodName) |
| 154 | + span.SetAttributes(qconfCallValueAttr) |
| 155 | + //requestCounter.Add(ctx, 1, metric.WithAttributes(qconfCallValueAttr)) |
| 156 | + |
| 157 | + results := method.Call(args) |
| 158 | + if len(results) > 1 { |
| 159 | + if err, ok := results[1].Interface().(error); ok && err != nil { |
| 160 | + logErr := fmt.Errorf("method call %s failed: %w", req.MethodName, |
| 161 | + results[0].Interface().(error)) |
| 162 | + a.fail(ctx, w, r, http.StatusInternalServerError, logErr.Error(), err) |
| 163 | + return |
| 164 | + } |
| 165 | + } |
| 166 | + |
| 167 | + if len(results) > 0 { |
| 168 | + // check if the result is an error |
| 169 | + if _, ok := results[0].Interface().(error); ok { |
| 170 | + logErr := fmt.Errorf("method call %s failed: %w", req.MethodName, |
| 171 | + results[0].Interface().(error)) |
| 172 | + a.fail(ctx, w, r, http.StatusInternalServerError, logErr.Error(), |
| 173 | + results[0].Interface().(error)) |
| 174 | + return |
| 175 | + } |
| 176 | + |
| 177 | + w.Header().Set("Content-Type", "application/json") |
| 178 | + if err := json.NewEncoder(w).Encode(results[0].Interface()); err != nil { |
| 179 | + logErr := fmt.Errorf("failed to encode response for method %s: %w", |
| 180 | + req.MethodName, err) |
| 181 | + a.fail(ctx, w, r, http.StatusInternalServerError, |
| 182 | + logErr.Error(), err) |
| 183 | + return |
| 184 | + } |
| 185 | + } |
| 186 | + logger.InfoContext(ctx, "request successfully processed", req.MethodName) |
| 187 | +} |
| 188 | + |
| 189 | +func (a *adapter) fail(ctx context.Context, w http.ResponseWriter, r *http.Request, status int, message string, err error) { |
| 190 | + w.WriteHeader(status) |
| 191 | + response := map[string]string{"error": message} |
| 192 | + logger.InfoContext(ctx, message, "URL", r.URL.Path) |
| 193 | + if err := json.NewEncoder(w).Encode(response); err != nil { |
| 194 | + logger.ErrorContext(ctx, "Failed to encode error response", err) |
| 195 | + } |
| 196 | + // write the error to the response body |
| 197 | + w.Write([]byte(message)) |
| 198 | +} |
| 199 | + |
| 200 | +func newLoggerProvider() (*log.LoggerProvider, error) { |
| 201 | + logExporter, err := stdoutlog.New() |
| 202 | + if err != nil { |
| 203 | + return nil, err |
| 204 | + } |
| 205 | + |
| 206 | + loggerProvider := log.NewLoggerProvider( |
| 207 | + log.WithProcessor(log.NewBatchProcessor(logExporter)), |
| 208 | + ) |
| 209 | + return loggerProvider, nil |
| 210 | +} |
| 211 | + |
| 212 | +func newTraceProvider() (*trace.TracerProvider, error) { |
| 213 | + traceExporter, err := stdouttrace.New( |
| 214 | + stdouttrace.WithPrettyPrint()) |
| 215 | + if err != nil { |
| 216 | + return nil, err |
| 217 | + } |
| 218 | + |
| 219 | + traceProvider := trace.NewTracerProvider( |
| 220 | + trace.WithBatcher(traceExporter, |
| 221 | + // Default is 5s. Set to 1s for demonstrative purposes. |
| 222 | + trace.WithBatchTimeout(time.Second)), |
| 223 | + ) |
| 224 | + return traceProvider, nil |
| 225 | +} |
| 226 | + |
| 227 | +func newMeterProvider() (*metric.MeterProvider, error) { |
| 228 | + metricExporter, err := stdoutmetric.New() |
| 229 | + if err != nil { |
| 230 | + return nil, err |
| 231 | + } |
| 232 | + |
| 233 | + meterProvider := metric.NewMeterProvider( |
| 234 | + metric.WithReader(metric.NewPeriodicReader(metricExporter, |
| 235 | + // Default is 1m. Set to 10s for demonstrative purposes. |
| 236 | + metric.WithInterval(10*time.Second))), |
| 237 | + ) |
| 238 | + return meterProvider, nil |
| 239 | +} |
0 commit comments