fileservice: refactor http trace

pkg/fileservice/http_client.go

@@ -37,10 +37,14 @@ var (
 )
 
 var dnsResolver = dns.NewCachingResolver(
-	net.DefaultResolver,
+	nil,
 	dns.MaxCacheEntries(128),
 )
 
+func init() {
+	net.DefaultResolver = dnsResolver
+}
+
 var httpDialer = &net.Dialer{
 	Timeout:  connectTimeout,
 	Resolver: dnsResolver,

pkg/fileservice/http_trace.go

@@ -0,0 +1,122 @@
+// Copyright 2024 Matrix Origin
+//
+// 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 fileservice
+
+import (
+	"crypto/tls"
+	"net/http/httptrace"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/common/reuse"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
+	metric "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
+	"go.uber.org/zap"
+)
+
+type traceInfo struct {
+	times traceTimes
+	trace *httptrace.ClientTrace
+}
+
+type traceTimes struct {
+	GetConn           time.Time
+	GotConn           time.Time
+	DNSStart          time.Time
+	ConnectStart      time.Time
+	TSLHandshakeStart time.Time
+}
+
+func newTraceInfo() *traceInfo {
+	info := new(traceInfo)
+	info.trace = &httptrace.ClientTrace{
+		GetConn:              info.GetConn,
+		GotConn:              info.GotConn,
+		PutIdleConn:          info.PutIdleConn,
+		GotFirstResponseByte: info.GotFirstResponseByte,
+		DNSStart:             info.DNSStart,
+		DNSDone:              info.DNSDone,
+		ConnectStart:         info.ConnectStart,
+		ConnectDone:          info.ConnectDone,
+		TLSHandshakeStart:    info.TLSHandshakeStart,
+		TLSHandshakeDone:     info.TLSHandshakeDone,
+	}
+	return info
+}
+
+func init() {
+	reuse.CreatePool(
+		newTraceInfo,
+		resetTracePoint,
+		reuse.DefaultOptions[traceInfo]().
+			WithEnableChecker())
+}
+
+func (traceInfo) TypeName() string {
+	return "fileservice.traceInfo"
+}
+
+func resetTracePoint(info *traceInfo) {
+	info.times = traceTimes{}
+}
+
+func (t *traceInfo) GetConn(hostPort string) {
+	t.times.GetConn = time.Now()
+	metric.FSHTTPTraceCounter.WithLabelValues("GetConn").Inc()
+}
+
+func (t *traceInfo) GotConn(info httptrace.GotConnInfo) {
+	t.times.GotConn = time.Now()
+	metric.FSHTTPTraceCounter.WithLabelValues("GotConn").Inc()
+	metric.FSHTTPTraceCounter.WithLabelValues("GotConnReused").Inc()
+	metric.FSHTTPTraceCounter.WithLabelValues("GotConnIdle").Inc()
+	metric.S3GetConnDurationHistogram.Observe(time.Since(t.times.GetConn).Seconds())
+}
+
+func (t *traceInfo) PutIdleConn(err error) {
+	logutil.Info("PutIdleConn error",
+		zap.Error(err),
+	)
+}
+
+func (t *traceInfo) GotFirstResponseByte() {
+	metric.S3GotFirstResponseDurationHistogram.Observe(time.Since(t.times.GotConn).Seconds())
+}
+
+func (t *traceInfo) DNSStart(di httptrace.DNSStartInfo) {
+	t.times.DNSStart = time.Now()
+	metric.FSHTTPTraceCounter.WithLabelValues("DNSStart").Inc()
+}
+
+func (t *traceInfo) DNSDone(di httptrace.DNSDoneInfo) {
+	metric.S3DNSResolveDurationHistogram.Observe(time.Since(t.times.DNSStart).Seconds())
+}
+
+func (t *traceInfo) ConnectStart(network, addr string) {
+	t.times.ConnectStart = time.Now()
+	metric.FSHTTPTraceCounter.WithLabelValues("ConnectStart").Inc()
+}
+
+func (t *traceInfo) ConnectDone(network, addr string, err error) {
+	metric.S3ConnectDurationHistogram.Observe(time.Since(t.times.ConnectStart).Seconds())
+}
+
+func (t *traceInfo) TLSHandshakeStart() {
+	metric.FSHTTPTraceCounter.WithLabelValues("TLSHandshakeStart").Inc()
+	t.times.TSLHandshakeStart = time.Now()
+}
+
+func (t *traceInfo) TLSHandshakeDone(cs tls.ConnectionState, err error) {
+	metric.S3TLSHandshakeDurationHistogram.Observe(time.Since(t.times.TSLHandshakeStart).Seconds())
+}

pkg/fileservice/object_storage_http_trace.go

@@ -0,0 +1,87 @@
+// Copyright 2024 Matrix Origin
+//
+// 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 fileservice
+
+import (
+	"context"
+	"io"
+	"iter"
+	"net/http/httptrace"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/common/reuse"
+)
+
+type objectStorageHTTPTrace struct {
+	upstream ObjectStorage
+}
+
+func newObjectStorageHTTPTrace(upstream ObjectStorage) *objectStorageHTTPTrace {
+	return &objectStorageHTTPTrace{
+		upstream: upstream,
+	}
+}
+
+var _ ObjectStorage = new(objectStorageHTTPTrace)
+
+func (o *objectStorageHTTPTrace) Delete(ctx context.Context, keys ...string) (err error) {
+	traceInfo := o.newTraceInfo()
+	defer o.closeTraceInfo(traceInfo)
+	ctx = httptrace.WithClientTrace(ctx, traceInfo.trace)
+	return o.upstream.Delete(ctx, keys...)
+}
+
+func (o *objectStorageHTTPTrace) Exists(ctx context.Context, key string) (bool, error) {
+	traceInfo := o.newTraceInfo()
+	defer o.closeTraceInfo(traceInfo)
+	ctx = httptrace.WithClientTrace(ctx, traceInfo.trace)
+	return o.upstream.Exists(ctx, key)
+}
+
+func (o *objectStorageHTTPTrace) List(ctx context.Context, prefix string) iter.Seq2[*DirEntry, error] {
+	traceInfo := o.newTraceInfo()
+	defer o.closeTraceInfo(traceInfo)
+	ctx = httptrace.WithClientTrace(ctx, traceInfo.trace)
+	return o.upstream.List(ctx, prefix)
+}
+
+func (o *objectStorageHTTPTrace) Read(ctx context.Context, key string, min *int64, max *int64) (r io.ReadCloser, err error) {
+	traceInfo := o.newTraceInfo()
+	defer o.closeTraceInfo(traceInfo)
+	ctx = httptrace.WithClientTrace(ctx, traceInfo.trace)
+	return o.upstream.Read(ctx, key, min, max)
+}
+
+func (o *objectStorageHTTPTrace) Stat(ctx context.Context, key string) (size int64, err error) {
+	traceInfo := o.newTraceInfo()
+	defer o.closeTraceInfo(traceInfo)
+	ctx = httptrace.WithClientTrace(ctx, traceInfo.trace)
+	return o.upstream.Stat(ctx, key)
+}
+
+func (o *objectStorageHTTPTrace) Write(ctx context.Context, key string, r io.Reader, size int64, expire *time.Time) (err error) {
+	traceInfo := o.newTraceInfo()
+	defer o.closeTraceInfo(traceInfo)
+	ctx = httptrace.WithClientTrace(ctx, traceInfo.trace)
+	return o.upstream.Write(ctx, key, r, size, expire)
+}
+
+func (o *objectStorageHTTPTrace) newTraceInfo() *traceInfo {
+	return reuse.Alloc[traceInfo](nil)
+}
+
+func (o *objectStorageHTTPTrace) closeTraceInfo(info *traceInfo) {
+	reuse.Free(info, nil)
+}

pkg/fileservice/s3_fs.go

@@ -17,11 +17,9 @@ package fileservice
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"errors"
 	"io"
 	"iter"
-	"net/http/httptrace"
 	pathpkg "path"
 	"runtime"
 	"sort"
@@ -30,7 +28,6 @@ import (
 	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
-	"github.com/matrixorigin/matrixone/pkg/common/reuse"
 	"github.com/matrixorigin/matrixone/pkg/fileservice/fscache"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/perfcounter"
@@ -138,6 +135,9 @@ func NewS3FS(
 		"s3",
 	)
 
+	// http trace
+	fs.storage = newObjectStorageHTTPTrace(fs.storage)
+
 	// cache
 	if !noCache {
 		if err := fs.initCaches(ctx, cacheConfig); err != nil {
@@ -370,10 +370,6 @@ func (s *S3FS) Write(ctx context.Context, vector IOVector) (err error) {
 		return err
 	}
 
-	tp := reuse.Alloc[tracePoint](nil)
-	defer reuse.Free(tp, nil)
-	ctx = httptrace.WithClientTrace(ctx, tp.getClientTrace())
-
 	var bytesWritten int
 	start := time.Now()
 	defer func() {
@@ -477,10 +473,6 @@ func (s *S3FS) Read(ctx context.Context, vector *IOVector) (err error) {
 		LogSlowEvent(ctx, time.Millisecond*500)
 	}()
 
-	tp := reuse.Alloc[tracePoint](nil)
-	defer reuse.Free(tp, nil)
-	ctx = httptrace.WithClientTrace(ctx, tp.getClientTrace())
-
 	if len(vector.Entries) == 0 {
 		return moerr.NewEmptyVectorNoCtx()
 	}
@@ -972,75 +964,3 @@ func (s *S3FS) Cost() *CostAttr {
 		List: CostHigh,
 	}
 }
-
-type tracePoint struct {
-	start             time.Time
-	dnsStart          time.Time
-	connectStart      time.Time
-	tlsHandshakeStart time.Time
-	ct                *httptrace.ClientTrace
-}
-
-func newTracePoint() *tracePoint {
-	tp := &tracePoint{
-		ct: &httptrace.ClientTrace{},
-	}
-	tp.ct.GetConn = tp.getConnPoint
-	tp.ct.GotConn = tp.gotConnPoint
-	tp.ct.DNSStart = tp.dnsStartPoint
-	tp.ct.DNSDone = tp.dnsDonePoint
-	tp.ct.ConnectStart = tp.connectStartPoint
-	tp.ct.ConnectDone = tp.connectDonePoint
-	tp.ct.TLSHandshakeStart = tp.tlsHandshakeStartPoint
-	tp.ct.TLSHandshakeDone = tp.tlsHandshakeDonePoint
-	return tp
-}
-
-func (tp tracePoint) TypeName() string {
-	return "fileservice.tracePoint"
-}
-
-func resetTracePoint(tp *tracePoint) {
-	tp.start = time.Time{}
-	tp.dnsStart = time.Time{}
-	tp.connectStart = time.Time{}
-	tp.tlsHandshakeStart = time.Time{}
-}
-
-func (tp *tracePoint) getClientTrace() *httptrace.ClientTrace {
-	return tp.ct
-}
-
-func (tp *tracePoint) getConnPoint(hostPort string) {
-	tp.start = time.Now()
-}
-
-func (tp *tracePoint) gotConnPoint(info httptrace.GotConnInfo) {
-	metric.S3GetConnDurationHistogram.Observe(time.Since(tp.start).Seconds())
-}
-
-func (tp *tracePoint) dnsStartPoint(di httptrace.DNSStartInfo) {
-	metric.S3DNSResolveCounter.Inc()
-	tp.dnsStart = time.Now()
-}
-
-func (tp *tracePoint) dnsDonePoint(di httptrace.DNSDoneInfo) {
-	metric.S3DNSResolveDurationHistogram.Observe(time.Since(tp.dnsStart).Seconds())
-}
-
-func (tp *tracePoint) connectStartPoint(network, addr string) {
-	metric.S3ConnectCounter.Inc()
-	tp.connectStart = time.Now()
-}
-
-func (tp *tracePoint) connectDonePoint(network, addr string, err error) {
-	metric.S3ConnectDurationHistogram.Observe(time.Since(tp.connectStart).Seconds())
-}
-
-func (tp *tracePoint) tlsHandshakeStartPoint() {
-	tp.tlsHandshakeStart = time.Now()
-}
-
-func (tp *tracePoint) tlsHandshakeDonePoint(cs tls.ConnectionState, err error) {
-	metric.S3TLSHandshakeDurationHistogram.Observe(time.Since(tp.tlsHandshakeStart).Seconds())
-}