Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 address_test.go //go:build go1.7 // +build go1.7 package csm import "testing" func TestAddressWithDefaults(t testing.T) { cases := map[string]struct { Host, Port string Expect string }{ "ip": { Host: "127.0.0.2", Port: "", Expect: "127.0.0.2:31000", }, "localhost": { Host: "localhost", Port: "", Expect: "127.0.0.1:31000", }, "uppercase localhost": { Host: "LOCALHOST", Port: "", Expect: "127.0.0.1:31000", }, "port": { Host: "localhost", Port: "32000", Expect: "127.0.0.1:32000", }, "ip6": { Host: "::1", Port: "", Expect: "[::1]:31000", }, "unset": { Host: "", Port: "", Expect: "127.0.0.1:31000", }, } for name, c := range cases { t.Run(name, func(t testing.T) { actual := AddressWithDefaults(c.Host, c.Port) if e, a := c.Expect, actual; e != a { t.Errorf("expect %v, got %v", e, a) } }) } }	858 B	2025-05-25 12:27
📄 doc.go // Package csm provides the Client Side Monitoring (CSM) client which enables // sending metrics via UDP connection to the CSM agent. This package provides // control options, and configuration for the CSM client. The client can be // controlled manually, or automatically via the SDK's Session configuration. // // Enabling CSM client via SDK's Session configuration // // The CSM client can be enabled automatically via SDK's Session configuration. // The SDK's session configuration enables the CSM client if the AWS_CSM_PORT // environment variable is set to a non-empty value. // // The configuration options for the CSM client via the SDK's session // configuration are: // // * AWS_CSM_PORT=<port number> // The port number the CSM agent will receive metrics on. // // * AWS_CSM_HOST=<hostname or ip> // The hostname, or IP address the CSM agent will receive metrics on. // Without port number. // // Manually enabling the CSM client // // The CSM client can be started, paused, and resumed manually. The Start // function will enable the CSM client to publish metrics to the CSM agent. It // is safe to call Start concurrently, but if Start is called additional times // with different ClientID or address it will panic. // // r, err := csm.Start("clientID", ":31000") // if err != nil { // panic(fmt.Errorf("failed starting CSM: %v", err)) // } // // When controlling the CSM client manually, you must also inject its request // handlers into the SDK's Session configuration for the SDK's API clients to // publish metrics. // // sess, err := session.NewSession(&aws.Config{}) // if err != nil { // panic(fmt.Errorf("failed loading session: %v", err)) // } // // // Add CSM client's metric publishing request handlers to the SDK's // // Session Configuration. // r.InjectHandlers(&sess.Handlers) // // Controlling CSM client // // Once the CSM client has been enabled the Get function will return a Reporter // value that you can use to pause and resume the metrics published to the CSM // agent. If Get function is called before the reporter is enabled with the // Start function or via SDK's Session configuration nil will be returned. // // The Pause method can be called to stop the CSM client publishing metrics to // the CSM agent. The Continue method will resume metric publishing. // // // Get the CSM client Reporter. // r := csm.Get() // // // Will pause monitoring // r.Pause() // resp, err = client.GetObject(&s3.GetObjectInput{ // Bucket: aws.String("bucket"), // Key: aws.String("key"), // }) // // // Resume monitoring // r.Continue() package csm	2.54 KB	2025-05-25 12:27
📄 enable.go package csm import ( "fmt" "strings" "sync" ) var ( lock sync.Mutex ) const ( // DefaultPort is used when no port is specified. DefaultPort = "31000" // DefaultHost is the host that will be used when none is specified. DefaultHost = "127.0.0.1" ) // AddressWithDefaults returns a CSM address built from the host and port // values. If the host or port is not set, default values will be used // instead. If host is "localhost" it will be replaced with "127.0.0.1". func AddressWithDefaults(host, port string) string { if len(host) == 0 \|\| strings.EqualFold(host, "localhost") { host = DefaultHost } if len(port) == 0 { port = DefaultPort } // Only IP6 host can contain a colon if strings.Contains(host, ":") { return "[" + host + "]:" + port } return host + ":" + port } // Start will start a long running go routine to capture // client side metrics. Calling start multiple time will only // start the metric listener once and will panic if a different // client ID or port is passed in. // // r, err := csm.Start("clientID", "127.0.0.1:31000") // if err != nil { // panic(fmt.Errorf("expected no error, but received %v", err)) // } // sess := session.NewSession() // r.InjectHandlers(sess.Handlers) // // svc := s3.New(sess) // out, err := svc.GetObject(&s3.GetObjectInput{ // Bucket: aws.String("bucket"), // Key: aws.String("key"), // }) func Start(clientID string, url string) (Reporter, error) { lock.Lock() defer lock.Unlock() if sender == nil { sender = newReporter(clientID, url) } else { if sender.clientID != clientID { panic(fmt.Errorf("inconsistent client IDs. %q was expected, but received %q", sender.clientID, clientID)) } if sender.url != url { panic(fmt.Errorf("inconsistent URLs. %q was expected, but received %q", sender.url, url)) } } if err := connect(url); err != nil { sender = nil return nil, err } return sender, nil } // Get will return a reporter if one exists, if one does not exist, nil will // be returned. func Get() Reporter { lock.Lock() defer lock.Unlock() return sender }	2.04 KB	2025-05-25 12:27
📄 enable_test.go package csm import ( "encoding/json" "fmt" "net" "testing" ) func startUDPServer(done chan struct{}, fn func([]byte)) (string, error) { addr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0") if err != nil { return "", err } conn, err := net.ListenUDP("udp", addr) if err != nil { return "", err } buf := make([]byte, 1024) go func() { defer conn.Close() for { select { case <-done: return default: } n, _, err := conn.ReadFromUDP(buf) fn(buf[:n]) if err != nil { panic(err) } } }() return conn.LocalAddr().String(), nil } func TestDifferentParams(t *testing.T) { defer func() { if r := recover(); r == nil { t.Errorf("expected panic with different parameters") } }() Start("clientID2", ":0") } var MetricsCh = make(chan map[string]interface{}, 1) var Done = make(chan struct{}) func init() { url, err := startUDPServer(Done, func(b []byte) { m := map[string]interface{}{} if err := json.Unmarshal(b, &m); err != nil { panic(fmt.Sprintf("expected no error, but received %v", err)) } MetricsCh <- m }) if err != nil { panic(err) } _, err = Start("clientID", url) if err != nil { panic(err) } }	1.17 KB	2025-05-25 12:27
📄 example_test.go package csm_test import ( "fmt" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/csm" "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/s3" ) func ExampleStart() { r, err := csm.Start("clientID", ":31000") if err != nil { panic(fmt.Errorf("failed starting CSM: %v", err)) } sess, err := session.NewSession(&aws.Config{}) if err != nil { panic(fmt.Errorf("failed loading session: %v", err)) } r.InjectHandlers(&sess.Handlers) client := s3.New(sess) client.GetObject(&s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) // Pauses monitoring r.Pause() client.GetObject(&s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) // Resume monitoring r.Continue() }	790 B	2025-05-25 12:27
📄 metric.go package csm import ( "strconv" "time" "github.com/aws/aws-sdk-go/aws" ) type metricTime time.Time func (t metricTime) MarshalJSON() ([]byte, error) { ns := time.Duration(time.Time(t).UnixNano()) return []byte(strconv.FormatInt(int64(ns/time.Millisecond), 10)), nil } type metric struct { ClientID string `json:"ClientId,omitempty"` API string `json:"Api,omitempty"` Service string `json:"Service,omitempty"` Timestamp metricTime `json:"Timestamp,omitempty"` Type string `json:"Type,omitempty"` Version int `json:"Version,omitempty"` AttemptCount int `json:"AttemptCount,omitempty"` Latency int `json:"Latency,omitempty"` Fqdn string `json:"Fqdn,omitempty"` UserAgent string `json:"UserAgent,omitempty"` AttemptLatency int `json:"AttemptLatency,omitempty"` SessionToken string `json:"SessionToken,omitempty"` Region string `json:"Region,omitempty"` AccessKey string `json:"AccessKey,omitempty"` HTTPStatusCode int `json:"HttpStatusCode,omitempty"` XAmzID2 string `json:"XAmzId2,omitempty"` XAmzRequestID string `json:"XAmznRequestId,omitempty"` AWSException string `json:"AwsException,omitempty"` AWSExceptionMessage string `json:"AwsExceptionMessage,omitempty"` SDKException string `json:"SdkException,omitempty"` SDKExceptionMessage string `json:"SdkExceptionMessage,omitempty"` FinalHTTPStatusCode int `json:"FinalHttpStatusCode,omitempty"` FinalAWSException string `json:"FinalAwsException,omitempty"` FinalAWSExceptionMessage string `json:"FinalAwsExceptionMessage,omitempty"` FinalSDKException string `json:"FinalSdkException,omitempty"` FinalSDKExceptionMessage string `json:"FinalSdkExceptionMessage,omitempty"` DestinationIP string `json:"DestinationIp,omitempty"` ConnectionReused int `json:"ConnectionReused,omitempty"` AcquireConnectionLatency int `json:"AcquireConnectionLatency,omitempty"` ConnectLatency int `json:"ConnectLatency,omitempty"` RequestLatency int `json:"RequestLatency,omitempty"` DNSLatency int `json:"DnsLatency,omitempty"` TCPLatency int `json:"TcpLatency,omitempty"` SSLLatency int `json:"SslLatency,omitempty"` MaxRetriesExceeded int `json:"MaxRetriesExceeded,omitempty"` } func (m metric) TruncateFields() { m.ClientID = truncateString(m.ClientID, 255) m.UserAgent = truncateString(m.UserAgent, 256) m.AWSException = truncateString(m.AWSException, 128) m.AWSExceptionMessage = truncateString(m.AWSExceptionMessage, 512) m.SDKException = truncateString(m.SDKException, 128) m.SDKExceptionMessage = truncateString(m.SDKExceptionMessage, 512) m.FinalAWSException = truncateString(m.FinalAWSException, 128) m.FinalAWSExceptionMessage = truncateString(m.FinalAWSExceptionMessage, 512) m.FinalSDKException = truncateString(m.FinalSDKException, 128) m.FinalSDKExceptionMessage = truncateString(m.FinalSDKExceptionMessage, 512) } func truncateString(v string, l int) string { if v != nil && len(v) > l { nv := (v)[:l] return &nv } return v } func (m metric) SetException(e metricException) { switch te := e.(type) { case awsException: m.AWSException = aws.String(te.exception) m.AWSExceptionMessage = aws.String(te.message) case sdkException: m.SDKException = aws.String(te.exception) m.SDKExceptionMessage = aws.String(te.message) } } func (m metric) SetFinalException(e metricException) { switch te := e.(type) { case awsException: m.FinalAWSException = aws.String(te.exception) m.FinalAWSExceptionMessage = aws.String(te.message) case sdkException: m.FinalSDKException = aws.String(te.exception) m.FinalSDKExceptionMessage = aws.String(te.message) } }	3.73 KB	2025-05-25 12:27
📄 metric_chan.go package csm import ( "sync/atomic" ) const ( runningEnum = iota pausedEnum ) var ( // MetricsChannelSize of metrics to hold in the channel MetricsChannelSize = 100 ) type metricChan struct { ch chan metric paused int64 } func newMetricChan(size int) metricChan { return metricChan{ ch: make(chan metric, size), paused: new(int64), } } func (ch metricChan) Pause() { atomic.StoreInt64(ch.paused, pausedEnum) } func (ch metricChan) Continue() { atomic.StoreInt64(ch.paused, runningEnum) } func (ch metricChan) IsPaused() bool { v := atomic.LoadInt64(ch.paused) return v == pausedEnum } // Push will push metrics to the metric channel if the channel // is not paused func (ch *metricChan) Push(m metric) bool { if ch.IsPaused() { return false } select { case ch.ch <- m: return true default: return false } }	857 B	2025-05-25 12:27
📄 metric_chan_test.go package csm import ( "testing" ) func TestMetricChanPush(t testing.T) { ch := newMetricChan(5) defer close(ch.ch) pushed := ch.Push(metric{}) if !pushed { t.Errorf("expected metrics to be pushed") } if e, a := 1, len(ch.ch); e != a { t.Errorf("expected %d, but received %d", e, a) } } func TestMetricChanPauseContinue(t testing.T) { ch := newMetricChan(5) defer close(ch.ch) ch.Pause() if !ch.IsPaused() { t.Errorf("expected to be paused, but did not pause properly") } ch.Continue() if ch.IsPaused() { t.Errorf("expected to be not paused, but did not continue properly") } pushed := ch.Push(metric{}) if !pushed { t.Errorf("expected metrics to be pushed") } if e, a := 1, len(ch.ch); e != a { t.Errorf("expected %d, but received %d", e, a) } } func TestMetricChanPushWhenPaused(t testing.T) { ch := newMetricChan(5) defer close(ch.ch) ch.Pause() pushed := ch.Push(metric{}) if pushed { t.Errorf("expected metrics to not be pushed") } if e, a := 0, len(ch.ch); e != a { t.Errorf("expected %d, but received %d", e, a) } } func TestMetricChanNonBlocking(t testing.T) { ch := newMetricChan(0) defer close(ch.ch) pushed := ch.Push(metric{}) if pushed { t.Errorf("expected metrics to be not pushed") } if e, a := 0, len(ch.ch); e != a { t.Errorf("expected %d, but received %d", e, a) } }	1.33 KB	2025-05-25 12:27
📄 metric_exception.go package csm type metricException interface { Exception() string Message() string } type requestException struct { exception string message string } func (e requestException) Exception() string { return e.exception } func (e requestException) Message() string { return e.message } type awsException struct { requestException } type sdkException struct { requestException }	387 B	2025-05-25 12:27
📄 metric_test.go //go:build go1.7 // +build go1.7 package csm import ( "reflect" "testing" "github.com/aws/aws-sdk-go/aws" ) func TestTruncateString(t testing.T) { cases := map[string]struct { Val string Len int Expect string }{ "no change": { Val: "123456789", Len: 10, Expect: "123456789", }, "max len": { Val: "1234567890", Len: 10, Expect: "1234567890", }, "too long": { Val: "12345678901", Len: 10, Expect: "1234567890", }, } for name, c := range cases { t.Run(name, func(t testing.T) { v := c.Val actual := truncateString(&v, c.Len) if e, a := c.Val, v; e != a { t.Errorf("expect input value not to change, %v, %v", e, a) } if e, a := c.Expect, actual; e != a { t.Errorf("expect %v, got %v", e, a) } }) } } func TestMetric_SetException(t testing.T) { cases := map[string]struct { Exc metricException Expect metric Final bool }{ "aws exc": { Exc: awsException{ requestException{exception: "abc", message: "123"}, }, Expect: metric{ AWSException: aws.String("abc"), AWSExceptionMessage: aws.String("123"), }, }, "sdk exc": { Exc: sdkException{ requestException{exception: "abc", message: "123"}, }, Expect: metric{ SDKException: aws.String("abc"), SDKExceptionMessage: aws.String("123"), }, }, "final aws exc": { Exc: awsException{ requestException{exception: "abc", message: "123"}, }, Expect: metric{ FinalAWSException: aws.String("abc"), FinalAWSExceptionMessage: aws.String("123"), }, Final: true, }, "final sdk exc": { Exc: sdkException{ requestException{exception: "abc", message: "123"}, }, Expect: metric{ FinalSDKException: aws.String("abc"), FinalSDKExceptionMessage: aws.String("123"), }, Final: true, }, } for name, c := range cases { t.Run(name, func(t *testing.T) { var m metric if c.Final { m.SetFinalException(c.Exc) } else { m.SetException(c.Exc) } if e, a := c.Expect, m; !reflect.DeepEqual(e, a) { t.Errorf("expect:\n%#v\nactual:\n%#v\n", e, a) } }) } }	2.1 KB	2025-05-25 12:27
📄 reporter.go package csm import ( "encoding/json" "net" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/request" ) // Reporter will gather metrics of API requests made and // send those metrics to the CSM endpoint. type Reporter struct { clientID string url string conn net.Conn metricsCh metricChan done chan struct{} } var ( sender Reporter ) func connect(url string) error { const network = "udp" if err := sender.connect(network, url); err != nil { return err } if sender.done == nil { sender.done = make(chan struct{}) go sender.start() } return nil } func newReporter(clientID, url string) Reporter { return &Reporter{ clientID: clientID, url: url, metricsCh: newMetricChan(MetricsChannelSize), } } func (rep Reporter) sendAPICallAttemptMetric(r request.Request) { if rep == nil { return } now := time.Now() creds, _ := r.Config.Credentials.Get() m := metric{ ClientID: aws.String(rep.clientID), API: aws.String(r.Operation.Name), Service: aws.String(r.ClientInfo.ServiceID), Timestamp: (metricTime)(&now), UserAgent: aws.String(r.HTTPRequest.Header.Get("User-Agent")), Region: r.Config.Region, Type: aws.String("ApiCallAttempt"), Version: aws.Int(1), XAmzRequestID: aws.String(r.RequestID), AttemptLatency: aws.Int(int(now.Sub(r.AttemptTime).Nanoseconds() / int64(time.Millisecond))), AccessKey: aws.String(creds.AccessKeyID), } if r.HTTPResponse != nil { m.HTTPStatusCode = aws.Int(r.HTTPResponse.StatusCode) } if r.Error != nil { if awserr, ok := r.Error.(awserr.Error); ok { m.SetException(getMetricException(awserr)) } } m.TruncateFields() rep.metricsCh.Push(m) } func getMetricException(err awserr.Error) metricException { msg := err.Error() code := err.Code() switch code { case request.ErrCodeRequestError, request.ErrCodeSerialization, request.CanceledErrorCode: return sdkException{ requestException{exception: code, message: msg}, } default: return awsException{ requestException{exception: code, message: msg}, } } } func (rep Reporter) sendAPICallMetric(r request.Request) { if rep == nil { return } now := time.Now() m := metric{ ClientID: aws.String(rep.clientID), API: aws.String(r.Operation.Name), Service: aws.String(r.ClientInfo.ServiceID), Timestamp: (metricTime)(&now), UserAgent: aws.String(r.HTTPRequest.Header.Get("User-Agent")), Type: aws.String("ApiCall"), AttemptCount: aws.Int(r.RetryCount + 1), Region: r.Config.Region, Latency: aws.Int(int(time.Since(r.Time) / time.Millisecond)), XAmzRequestID: aws.String(r.RequestID), MaxRetriesExceeded: aws.Int(boolIntValue(r.RetryCount >= r.MaxRetries())), } if r.HTTPResponse != nil { m.FinalHTTPStatusCode = aws.Int(r.HTTPResponse.StatusCode) } if r.Error != nil { if awserr, ok := r.Error.(awserr.Error); ok { m.SetFinalException(getMetricException(awserr)) } } m.TruncateFields() // TODO: Probably want to figure something out for logging dropped // metrics rep.metricsCh.Push(m) } func (rep Reporter) connect(network, url string) error { if rep.conn != nil { rep.conn.Close() } conn, err := net.Dial(network, url) if err != nil { return awserr.New("UDPError", "Could not connect", err) } rep.conn = conn return nil } func (rep Reporter) close() { if rep.done != nil { close(rep.done) } rep.metricsCh.Pause() } func (rep Reporter) start() { defer func() { rep.metricsCh.Pause() }() for { select { case <-rep.done: rep.done = nil return case m := <-rep.metricsCh.ch: // TODO: What to do with this error? Probably should just log b, err := json.Marshal(m) if err != nil { continue } rep.conn.Write(b) } } } // Pause will pause the metric channel preventing any new metrics from being // added. It is safe to call concurrently with other calls to Pause, but if // called concurently with Continue can lead to unexpected state. func (rep Reporter) Pause() { lock.Lock() defer lock.Unlock() if rep == nil { return } rep.close() } // Continue will reopen the metric channel and allow for monitoring to be // resumed. It is safe to call concurrently with other calls to Continue, but // if called concurently with Pause can lead to unexpected state. func (rep Reporter) Continue() { lock.Lock() defer lock.Unlock() if rep == nil { return } if !rep.metricsCh.IsPaused() { return } rep.metricsCh.Continue() } // Client side metric handler names const ( APICallMetricHandlerName = "awscsm.SendAPICallMetric" APICallAttemptMetricHandlerName = "awscsm.SendAPICallAttemptMetric" ) // InjectHandlers will will enable client side metrics and inject the proper // handlers to handle how metrics are sent. // // InjectHandlers is NOT safe to call concurrently. Calling InjectHandlers // multiple times may lead to unexpected behavior, (e.g. duplicate metrics). // // // Start must be called in order to inject the correct handlers // r, err := csm.Start("clientID", "127.0.0.1:8094") // if err != nil { // panic(fmt.Errorf("expected no error, but received %v", err)) // } // // sess := session.NewSession() // r.InjectHandlers(&sess.Handlers) // // // create a new service client with our client side metric session // svc := s3.New(sess) func (rep Reporter) InjectHandlers(handlers *request.Handlers) { if rep == nil { return } handlers.Complete.PushFrontNamed(request.NamedHandler{ Name: APICallMetricHandlerName, Fn: rep.sendAPICallMetric, }) handlers.CompleteAttempt.PushFrontNamed(request.NamedHandler{ Name: APICallAttemptMetricHandlerName, Fn: rep.sendAPICallAttemptMetric, }) } // boolIntValue return 1 for true and 0 for false. func boolIntValue(b bool) int { if b { return 1 } return 0 }	5.84 KB	2025-05-25 12:27
📄 reporter_internal_test.go package csm import ( "net/http" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/client" "github.com/aws/aws-sdk-go/aws/client/metadata" "github.com/aws/aws-sdk-go/aws/credentials" "github.com/aws/aws-sdk-go/aws/defaults" "github.com/aws/aws-sdk-go/aws/request" ) func TestMaxRetriesExceeded(t testing.T) { md := metadata.ClientInfo{ Endpoint: "http://127.0.0.1", } cfg := aws.Config{ Region: aws.String("foo"), Credentials: credentials.NewStaticCredentials("", "", ""), } op := &request.Operation{} cases := []struct { name string httpStatusCode int expectedMaxRetriesValue int expectedMetrics int }{ { name: "max retry reached", httpStatusCode: http.StatusBadGateway, expectedMaxRetriesValue: 1, }, { name: "status ok", httpStatusCode: http.StatusOK, expectedMaxRetriesValue: 0, }, } for _, c := range cases { r := request.New(cfg, md, defaults.Handlers(), client.DefaultRetryer{NumMaxRetries: 2}, op, nil, nil) reporter := newReporter("", "") r.Handlers.Send.Clear() reporter.InjectHandlers(&r.Handlers) r.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &http.Response{ StatusCode: c.httpStatusCode, } }) r.Send() for { m := <-reporter.metricsCh.ch if m.Type != "ApiCall" { // ignore non-ApiCall metrics since MaxRetriesExceeded is only on ApiCall events continue } if val := m.MaxRetriesExceeded; val != c.expectedMaxRetriesValue { t.Errorf("%s: expected %d, but received %d", c.name, c.expectedMaxRetriesValue, val) } break } } }	1.67 KB	2025-05-25 12:27
📄 reporter_test.go //go:build go1.7 // +build go1.7 package csm_test import ( "context" "fmt" "net/http" "net/http/httptest" "sort" "testing" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/client" "github.com/aws/aws-sdk-go/aws/client/metadata" "github.com/aws/aws-sdk-go/aws/csm" "github.com/aws/aws-sdk-go/aws/request" v4 "github.com/aws/aws-sdk-go/aws/signer/v4" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/private/protocol/jsonrpc" ) func TestReportingMetrics(t testing.T) { sess := unit.Session.Copy(&aws.Config{ SleepDelay: func(time.Duration) {}, }) sess.Handlers.Validate.Clear() sess.Handlers.Sign.Clear() sess.Handlers.Send.Clear() reporter := csm.Get() if reporter == nil { t.Errorf("expected non-nil reporter") } reporter.InjectHandlers(&sess.Handlers) cases := map[string]struct { Request request.Request ExpectMetrics []map[string]interface{} }{ "successful request": { Request: func() request.Request { md := metadata.ClientInfo{} op := &request.Operation{Name: "OperationName"} req := request.New(sess.Config, md, sess.Handlers, client.DefaultRetryer{NumMaxRetries: 3}, op, nil, nil) req.Handlers.Send.PushBack(func(r request.Request) { req.HTTPResponse = &http.Response{ StatusCode: 200, Header: http.Header{}, } }) return req }(), ExpectMetrics: []map[string]interface{}{ { "Type": "ApiCallAttempt", "HttpStatusCode": float64(200), }, { "Type": "ApiCall", "FinalHttpStatusCode": float64(200), }, }, }, "failed request, no retry": { Request: func() request.Request { md := metadata.ClientInfo{} op := &request.Operation{Name: "OperationName"} req := request.New(sess.Config, md, sess.Handlers, client.DefaultRetryer{NumMaxRetries: 3}, op, nil, nil) req.Handlers.Send.PushBack(func(r request.Request) { req.HTTPResponse = &http.Response{ StatusCode: 400, Header: http.Header{}, } req.Retryable = aws.Bool(false) req.Error = awserr.New("Error", "Message", nil) }) return req }(), ExpectMetrics: []map[string]interface{}{ { "Type": "ApiCallAttempt", "HttpStatusCode": float64(400), "AwsException": "Error", "AwsExceptionMessage": "Error: Message", }, { "Type": "ApiCall", "FinalHttpStatusCode": float64(400), "FinalAwsException": "Error", "FinalAwsExceptionMessage": "Error: Message", "AttemptCount": float64(1), }, }, }, "failed request, with retry": { Request: func() request.Request { md := metadata.ClientInfo{} op := &request.Operation{Name: "OperationName"} req := request.New(sess.Config, md, sess.Handlers, client.DefaultRetryer{NumMaxRetries: 1}, op, nil, nil) resps := []http.Response{ { StatusCode: 500, Header: http.Header{}, }, { StatusCode: 500, Header: http.Header{}, }, } req.Handlers.Send.PushBack(func(r request.Request) { req.HTTPResponse = resps[0] resps = resps[1:] }) return req }(), ExpectMetrics: []map[string]interface{}{ { "Type": "ApiCallAttempt", "HttpStatusCode": float64(500), "AwsException": "UnknownError", "AwsExceptionMessage": "UnknownError: unknown error", }, { "Type": "ApiCallAttempt", "HttpStatusCode": float64(500), "AwsException": "UnknownError", "AwsExceptionMessage": "UnknownError: unknown error", }, { "Type": "ApiCall", "FinalHttpStatusCode": float64(500), "FinalAwsException": "UnknownError", "FinalAwsExceptionMessage": "UnknownError: unknown error", "AttemptCount": float64(2), }, }, }, "success request, with retry": { Request: func() request.Request { md := metadata.ClientInfo{} op := &request.Operation{Name: "OperationName"} req := request.New(sess.Config, md, sess.Handlers, client.DefaultRetryer{NumMaxRetries: 3}, op, nil, nil) errs := []error{ awserr.New("AWSError", "aws error", nil), awserr.New(request.ErrCodeRequestError, "sdk error", nil), nil, } resps := []http.Response{ { StatusCode: 500, Header: http.Header{}, }, { StatusCode: 500, Header: http.Header{}, }, { StatusCode: 200, Header: http.Header{}, }, } req.Handlers.Send.PushBack(func(r request.Request) { req.HTTPResponse = resps[0] resps = resps[1:] req.Error = errs[0] errs = errs[1:] }) return req }(), ExpectMetrics: []map[string]interface{}{ { "Type": "ApiCallAttempt", "AwsException": "AWSError", "AwsExceptionMessage": "AWSError: aws error", "HttpStatusCode": float64(500), }, { "Type": "ApiCallAttempt", "SdkException": request.ErrCodeRequestError, "SdkExceptionMessage": request.ErrCodeRequestError + ": sdk error", "HttpStatusCode": float64(500), }, { "Type": "ApiCallAttempt", "AwsException": nil, "AwsExceptionMessage": nil, "SdkException": nil, "SdkExceptionMessage": nil, "HttpStatusCode": float64(200), }, { "Type": "ApiCall", "FinalHttpStatusCode": float64(200), "FinalAwsException": nil, "FinalAwsExceptionMessage": nil, "FinalSdkException": nil, "FinalSdkExceptionMessage": nil, "AttemptCount": float64(3), }, }, }, } for name, c := range cases { t.Run(name, func(t testing.T) { ctx, cancelFn := context.WithTimeout(context.Background(), time.Second) defer cancelFn() c.Request.Send() for i := 0; i < len(c.ExpectMetrics); i++ { select { case m := <-csm.MetricsCh: for ek, ev := range c.ExpectMetrics[i] { if ev == nil { // must not be set if _, ok := m[ek]; ok { t.Errorf("%d, expect %v metric member, not to be set, %v", i, ek, m[ek]) } continue } if _, ok := m[ek]; !ok { t.Errorf("%d, expect %v metric member, keys: %v", i, ek, keys(m)) } if e, a := ev, m[ek]; e != a { t.Errorf("%d, expect %v:%v(%T), metric value, got %v(%T)", i, ek, e, e, a, a) } } case <-ctx.Done(): t.Errorf("timeout waiting for metrics") return } } var extraMetrics []map[string]interface{} Loop: for { select { case m := <-csm.MetricsCh: extraMetrics = append(extraMetrics, m) default: break Loop } } if len(extraMetrics) != 0 { t.Fatalf("unexpected metrics, %#v", extraMetrics) } }) } } type mockService struct { client.Client } type input struct{} type output struct{} func (s mockService) Request(i input) request.Request { op := &request.Operation{ Name: "foo", HTTPMethod: "POST", HTTPPath: "/", } o := output{} req := s.NewRequest(op, &i, &o) return req } func BenchmarkWithCSM(b testing.B) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.Write([]byte(fmt.Sprintf("{}"))) })) defer server.Close() cfg := aws.Config{ Endpoint: aws.String(server.URL), } sess := unit.Session.Copy(&cfg) r := csm.Get() r.InjectHandlers(&sess.Handlers) c := sess.ClientConfig("id", &cfg) svc := mockService{ client.New( c.Config, metadata.ClientInfo{ ServiceName: "service", ServiceID: "id", SigningName: "signing", SigningRegion: "region", Endpoint: server.URL, APIVersion: "0", JSONVersion: "1.1", TargetPrefix: "prefix", }, c.Handlers, ), } svc.Handlers.Sign.PushBackNamed(v4.SignRequestHandler) svc.Handlers.Build.PushBackNamed(jsonrpc.BuildHandler) svc.Handlers.Unmarshal.PushBackNamed(jsonrpc.UnmarshalHandler) svc.Handlers.UnmarshalMeta.PushBackNamed(jsonrpc.UnmarshalMetaHandler) svc.Handlers.UnmarshalError.PushBackNamed(jsonrpc.UnmarshalErrorHandler) for i := 0; i < b.N; i++ { req := svc.Request(input{}) req.Send() } } func BenchmarkWithCSMNoUDPConnection(b testing.B) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.Write([]byte(fmt.Sprintf("{}"))) })) defer server.Close() cfg := aws.Config{ Endpoint: aws.String(server.URL), } sess := unit.Session.Copy(&cfg) r := csm.Get() r.Pause() r.InjectHandlers(&sess.Handlers) defer r.Pause() c := sess.ClientConfig("id", &cfg) svc := mockService{ client.New( c.Config, metadata.ClientInfo{ ServiceName: "service", ServiceID: "id", SigningName: "signing", SigningRegion: "region", Endpoint: server.URL, APIVersion: "0", JSONVersion: "1.1", TargetPrefix: "prefix", }, c.Handlers, ), } svc.Handlers.Sign.PushBackNamed(v4.SignRequestHandler) svc.Handlers.Build.PushBackNamed(jsonrpc.BuildHandler) svc.Handlers.Unmarshal.PushBackNamed(jsonrpc.UnmarshalHandler) svc.Handlers.UnmarshalMeta.PushBackNamed(jsonrpc.UnmarshalMetaHandler) svc.Handlers.UnmarshalError.PushBackNamed(jsonrpc.UnmarshalErrorHandler) for i := 0; i < b.N; i++ { req := svc.Request(input{}) req.Send() } } func BenchmarkWithoutCSM(b testing.B) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.Write([]byte(fmt.Sprintf("{}"))) })) defer server.Close() cfg := aws.Config{ Endpoint: aws.String(server.URL), } sess := unit.Session.Copy(&cfg) c := sess.ClientConfig("id", &cfg) svc := mockService{ client.New( *c.Config, metadata.ClientInfo{ ServiceName: "service", ServiceID: "id", SigningName: "signing", SigningRegion: "region", Endpoint: server.URL, APIVersion: "0", JSONVersion: "1.1", TargetPrefix: "prefix", }, c.Handlers, ), } svc.Handlers.Sign.PushBackNamed(v4.SignRequestHandler) svc.Handlers.Build.PushBackNamed(jsonrpc.BuildHandler) svc.Handlers.Unmarshal.PushBackNamed(jsonrpc.UnmarshalHandler) svc.Handlers.UnmarshalMeta.PushBackNamed(jsonrpc.UnmarshalMetaHandler) svc.Handlers.UnmarshalError.PushBackNamed(jsonrpc.UnmarshalErrorHandler) for i := 0; i < b.N; i++ { req := svc.Request(input{}) req.Send() } } func keys(m map[string]interface{}) []string { ks := make([]string, 0, len(m)) for k := range m { ks = append(ks, k) } sort.Strings(ks) return ks }	10.54 KB	2025-05-25 12:27

Terminal