Merge pull request #7 from JeffreyRichter/test-feature

Reworked default logging

Author: JeffreyRichter

2016-05-31/azblob/policy_request_log.go

@@ -15,7 +15,8 @@ import (
 
 // RequestLogOptions configures the retry policy's behavior.
 type RequestLogOptions struct {
-	// LogWarningIfTryOverThreshold logs a warning if a tried operation takes longer than the specified duration (0=no logging).
+	// LogWarningIfTryOverThreshold logs a warning if a tried operation takes longer than the specified
+	// duration (-1=no logging; 0=default threshold).
 	LogWarningIfTryOverThreshold time.Duration
 }
 
@@ -40,45 +41,50 @@ func (f *requestLogPolicyFactory) New(node pipeline.Node) pipeline.Policy {
 type requestLogPolicy struct {
 	node           pipeline.Node
 	o              RequestLogOptions
-	try            int
+	try            int32
 	operationStart time.Time
 }
 
 func redactSigQueryParam(rawQuery string) (bool, string) {
-	sigFound := strings.EqualFold(rawQuery, "?sig=")
+	rawQuery = strings.ToLower(rawQuery) // lowercase the string so we can look for ?sig= and &sig=
+	sigFound := strings.Contains(rawQuery, "?sig=")
 	if !sigFound {
-		sigFound = strings.EqualFold(rawQuery, "&sig=")
+		sigFound = strings.Contains(rawQuery, "&sig=")
 		if !sigFound {
 			return sigFound, rawQuery // [?|&]sig= not found; return same rawQuery passed in (no memory allocation)
 		}
 	}
-	// [?|&]sig= was found, redact its value
+	// [?|&]sig= found, redact its value
 	values, _ := url.ParseQuery(rawQuery)
 	for name := range values {
 		if strings.EqualFold(name, "sig") {
-			values[name] = []string{"(redacted)"}
+			values[name] = []string{"REDACTED"}
 		}
 	}
 	return sigFound, values.Encode()
 }
 
+func prepareRequestForLogging(request pipeline.Request) *http.Request {
+	req := request
+	if sigFound, rawQuery := redactSigQueryParam(req.URL.RawQuery); sigFound {
+		// Make copy so we don't destroy the query parameters we actually need to send in the request
+		req = request.Copy()
+		req.Request.URL.RawQuery = rawQuery
+	}
+	return req.Request
+}
+
 func (p *requestLogPolicy) Do(ctx context.Context, request pipeline.Request) (response pipeline.Response, err error) {
 	p.try++ // The first try is #1 (not #0)
 	if p.try == 1 {
 		p.operationStart = time.Now() // If this is the 1st try, record the operation state time
 	}
 
 	// Log the outgoing request as informational
-	if p.node.WouldLog(pipeline.LogInfo) {
+	if p.node.ShouldLog(pipeline.LogInfo) {
 		b := &bytes.Buffer{}
 		fmt.Fprintf(b, "==> OUTGOING REQUEST (Try=%d)\n", p.try)
-		req := request
-		if sigFound, rawQuery := redactSigQueryParam(req.URL.RawQuery); sigFound {
-			// TODO: Make copy so we dont' destroy the query parameters we actually need to send in the request
-			req = request.Copy()
-			req.Request.URL.RawQuery = rawQuery
-		}
-		pipeline.WriteRequest(b, req.Request)
+		pipeline.WriteRequest(b, prepareRequestForLogging(request))
 		p.node.Log(pipeline.LogInfo, b.String())
 	}
 
@@ -92,16 +98,18 @@ func (p *requestLogPolicy) Do(ctx context.Context, request pipeline.Request) (re
 	severity := pipeline.LogInfo // Assume success and default to informational logging
 	logMsg := func(b *bytes.Buffer) {
 		b.WriteString("SUCCESSFUL OPERATION\n")
-		pipeline.WriteResponseWithRequest(b, response.Response())
+		pipeline.WriteRequestWithResponse(b, prepareRequestForLogging(request), response.Response())
 	}
 
+	forceLog := false
 	// If the response took too long, we'll upgrade to warning.
 	if p.o.LogWarningIfTryOverThreshold > 0 && tryDuration > p.o.LogWarningIfTryOverThreshold {
 		// Log a warning if the try duration exceeded the specified threshold
 		severity = pipeline.LogWarning
 		logMsg = func(b *bytes.Buffer) {
 			fmt.Fprintf(b, "SLOW OPERATION [tryDuration > %v]\n", p.o.LogWarningIfTryOverThreshold)
-			pipeline.WriteResponseWithRequest(b, response.Response())
+			pipeline.WriteRequestWithResponse(b, prepareRequestForLogging(request), response.Response())
+			forceLog = true // For CSS (Customer Support Services), we always log these to help diagnose latency issues
 		}
 	}
 
@@ -111,9 +119,10 @@ func (p *requestLogPolicy) Do(ctx context.Context, request pipeline.Request) (re
 			severity = pipeline.LogError // Promote to Error any 4xx (except those listed is an error) or any 5xx
 			logMsg = func(b *bytes.Buffer) {
 				// Write the error, the originating request and the stack
-				fmt.Fprintf(b, "OPERATION ERROR:\n%v\n", err)
-				pipeline.WriteResponseWithRequest(b, response.Response())
+				fmt.Fprintf(b, "OPERATION ERROR:\n")
+				pipeline.WriteRequestWithResponse(b, prepareRequestForLogging(request), response.Response())
 				b.Write(stack()) // For errors, we append the stack trace (an expensive operation)
+				forceLog = true  // TODO: Do we really want this here?
 			}
 		} else {
 			// For other status codes, we leave the severity as is.
@@ -123,17 +132,25 @@ func (p *requestLogPolicy) Do(ctx context.Context, request pipeline.Request) (re
 		logMsg = func(b *bytes.Buffer) {
 			// Write the error, the originating request and the stack
 			fmt.Fprintf(b, "NETWORK ERROR:\n%v\n", err)
-			pipeline.WriteRequest(b, request.Request)
+			pipeline.WriteRequest(b, prepareRequestForLogging(request))
 			b.Write(stack()) // For errors, we append the stack trace (an expensive operation)
+			forceLog = true
 		}
 	}
 
-	if p.node.WouldLog(severity) || false { // Change false to true for testing
+	if shouldLog := p.node.ShouldLog(severity); forceLog || shouldLog {
 		// We're going to log this; build the string to log
 		b := &bytes.Buffer{}
 		fmt.Fprintf(b, "==> REQUEST/RESPONSE (Try=%d, TryDuration=%v, OpDuration=%v) -- ", p.try, tryDuration, opDuration)
 		logMsg(b)
-		p.node.Log(severity, b.String())
+		msg := b.String()
+
+		if forceLog {
+			pipeline.ForceLog(severity, msg)
+		}
+		if shouldLog {
+			p.node.Log(severity, msg)
+		}
 	}
 	return response, err
 }

2016-05-31/azblob/policy_retry.go

@@ -12,6 +12,9 @@ import (
 	"github.com/Azure/azure-pipeline-go/pipeline"
 )
 
+// RetryPolicy tells the pipeline what kind of retry policy to use. See the RetryPolicy* constants.
+type RetryPolicy int32
+
 const (
 	// RetryPolicyExponential tells the pipeline to use an exponential back-off retry policy
 	RetryPolicyExponential RetryPolicy = 0
@@ -20,9 +23,6 @@ const (
 	RetryPolicyFixed RetryPolicy = 1
 )
 
-// RetryPolicy tells the pipeline what kind of retry policy to use. See the RetryPolicy* constants.
-type RetryPolicy int
-
 // RetryOptions configures the retry policy's behavior.
 type RetryOptions struct {
 	// Policy tells the pipeline what kind of retry policy to use. See the RetryPolicy* constants.\
@@ -31,7 +31,7 @@ type RetryOptions struct {
 
 	// MaxTries specifies the maximum number of attempts an operation will be tried before producing an error (0=default).
 	// A value of zero means that you accept our default policy. A value of 1 means 1 try and no retries.
-	MaxTries int
+	MaxTries int32
 
 	// TryTimeout indicates the maximum time allowed for any single try of an HTTP request.
 	// A value of zero means that you accept our default timeout. NOTE: When transferring large amounts
@@ -85,10 +85,10 @@ func (o RetryOptions) defaults() RetryOptions {
 	return o
 }
 
-func (o RetryOptions) calcDelay(try int) time.Duration { // try is >=1; never 0
-	pow := func(number int64, exponent int) int64 { // pow is nested helper function
+func (o RetryOptions) calcDelay(try int32) time.Duration { // try is >=1; never 0
+	pow := func(number int64, exponent int32) int64 { // pow is nested helper function
 		var result int64 = 1
-		for n := 0; n < exponent; n++ {
+		for n := int32(0); n < exponent; n++ {
 			result *= number
 		}
 		return result
@@ -139,7 +139,7 @@ var logf = func(format string, a ...interface{}) {}
 func (p *retryPolicy) Do(ctx context.Context, request pipeline.Request) (response pipeline.Response, err error) {
 	// Before each try, we'll select either the primary or secondary URL.
 	secondaryHost := ""
-	primaryTry := 0 // This indicates how many tries we've attempted against the primary DC
+	primaryTry := int32(0) // This indicates how many tries we've attempted against the primary DC
 
 	// We only consider retring against a secondary if we have a read request (GET/HEAD) AND this policy has a Secondary URL it can use
 	considerSecondary := (request.Method == http.MethodGet || request.Method == http.MethodHead) && p.o.RetryReadsFromSecondaryHost != ""
@@ -154,7 +154,7 @@ func (p *retryPolicy) Do(ctx context.Context, request pipeline.Request) (respons
 	//    For a primary wait ((2 ^ primaryTries - 1) * delay * random(0.8, 1.2)
 	//    If secondary gets a 404, don't fail, retry but future retries are only against the primary
 	//    When retrying against a secondary, ignore the retry count and wait (.1 second * random(0.8, 1.2))
-	for try := 1; try <= p.o.MaxTries; try++ {
+	for try := int32(1); try <= p.o.MaxTries; try++ {
 		logf("\n=====> Try=%d\n", try)
 
 		// Determine which endpoint to try. It's primary if there is no secondary or if it is an add # attempt.
@@ -184,9 +184,9 @@ func (p *retryPolicy) Do(ctx context.Context, request pipeline.Request) (respons
 		}
 
 		// Set the server-side timeout query parameter "timeout=[seconds]"
-		timeout := int(p.o.TryTimeout.Seconds()) // Max seconds per try
-		if deadline, ok := ctx.Deadline(); ok {  // If user's ctx has a deadline, make the timeout the smaller of the two
-			t := int(deadline.Sub(time.Now()).Seconds()) // Duration from now until user's ctx reaches its deadline
+		timeout := int32(p.o.TryTimeout.Seconds()) // Max seconds per try
+		if deadline, ok := ctx.Deadline(); ok {    // If user's ctx has a deadline, make the timeout the smaller of the two
+			t := int32(deadline.Sub(time.Now()).Seconds()) // Duration from now until user's ctx reaches its deadline
 			logf("MaxTryTimeout=%d secs, TimeTilDeadline=%d sec\n", timeout, t)
 			if t < timeout {
 				timeout = t
@@ -197,7 +197,7 @@ func (p *retryPolicy) Do(ctx context.Context, request pipeline.Request) (respons
 			logf("TryTimeout adjusted to=%d sec\n", timeout)
 		}
 		q := requestCopy.Request.URL.Query()
-		q.Set("timeout", strconv.Itoa(timeout))
+		q.Set("timeout", strconv.Itoa(int(timeout)))
 		requestCopy.Request.URL.RawQuery = q.Encode()
 		logf("Url=%s\n", requestCopy.Request.URL.String())
 

2016-05-31/azblob/storage_error.go

@@ -68,7 +68,8 @@ func (e *storageError) Error() string {
 			fmt.Fprintf(b, "   %s: %+v\n", k, e.details[k])
 		}
 	}
-	pipeline.WriteResponseWithRequest(b, e.response)
+	req := pipeline.Request{Request: e.response.Request}.Copy() // Make a copy of the response's request
+	pipeline.WriteRequestWithResponse(b, prepareRequestForLogging(req), e.response)
 	return e.ErrorNode.Error(b.String())
 }
 

2016-05-31/azblob/zt_examples_test.go

@@ -79,6 +79,7 @@ func Example() {
 
 	downloadedData := &bytes.Buffer{}
 	downloadedData.ReadFrom(get.Body())
+	get.Body().Close() // The client must close the response body when finished with it
 	if data != downloadedData.String() {
 		log.Fatal("downloaded data doesn't match uploaded data")
 	}
@@ -431,6 +432,9 @@ func ExampleBlobAccessConditions() {
 				fmt.Print("Failure: " + serr.Response().Status + "\n")
 			}
 		} else {
+			if get, ok := response.(*GetResponse); ok {
+				get.Body().Close() // The client must close the response body when finished with it
+			}
 			fmt.Print("Success: " + response.Response().Status + "\n")
 		}
 	}
@@ -668,7 +672,7 @@ func ExampleBlockBlobURL() {
 	}
 	blobData := &bytes.Buffer{}
 	blobData.ReadFrom(get.Body())
-	get.Body().Close()
+	get.Body().Close() // The client must close the response body when finished with it
 	fmt.Println(blobData)
 }
 
@@ -703,7 +707,7 @@ func ExampleAppendBlobURL() {
 	}
 	b := bytes.Buffer{}
 	b.ReadFrom(get.Body())
-	get.Body().Close()
+	get.Body().Close() // The client must close the response body when finished with it
 	fmt.Println(b.String())
 }
 
@@ -765,7 +769,7 @@ func ExamplePageBlobURL() {
 	}
 	blobData := &bytes.Buffer{}
 	blobData.ReadFrom(get.Body())
-	get.Body().Close()
+	get.Body().Close() // The client must close the response body when finished with it
 	fmt.Printf("%#v", blobData.Bytes())
 }
 
@@ -804,15 +808,15 @@ func Example_blobSnapshots() {
 	get, err := baseBlobURL.GetBlob(ctx, BlobRange{}, BlobAccessConditions{}, false)
 	b := bytes.Buffer{}
 	b.ReadFrom(get.Body())
-	get.Body().Close()
+	get.Body().Close() // The client must close the response body when finished with it
 	fmt.Println(b.String())
 
 	// Show snapshot blob via original blob URI & snapshot time:
 	snapshotBlobURL := baseBlobURL.WithSnapshot(snapshot)
 	get, err = snapshotBlobURL.GetBlob(ctx, BlobRange{}, BlobAccessConditions{}, false)
 	b.Reset()
 	b.ReadFrom(get.Body())
-	get.Body().Close()
+	get.Body().Close() // The client must close the response body when finished with it
 	fmt.Println(b.String())
 
 	// FYI: You can get the base blob URL from one of its snapshot by passing time.Time{} to WithSnapshot:
@@ -901,6 +905,7 @@ func Example_progressUploadDownload() {
 
 	downloadedData := &bytes.Buffer{}
 	downloadedData.ReadFrom(responseBody)
+	responseBody.Close() // The client must close the response body when finished with it
 	// The downloaded blob data is in downloadData's buffer
 }
 
@@ -1008,7 +1013,7 @@ func ExampleNewGetRetryStream() {
 		func(bytesTransferred int64) {
 			fmt.Printf("Downloaded %d of %d bytes.\n", bytesTransferred, contentLength)
 		})
-	defer stream.Close()
+	defer stream.Close() // The client must close the response body when finished with it
 
 	file, err := os.Create("BigFile.bin") // Create the file to hold the downloaded blob contents.
 	if err != nil {

2016-05-31/azblob/zt_policy_retry_test.go

@@ -15,7 +15,7 @@ import (
 	"github.com/Azure/azure-storage-blob-go/2016-05-31/azblob"
 )
 
-type retryTestScenario int
+type retryTestScenario int32
 
 const (
 	// Retry until success. Max reties hit. Operation time out prevents additional retries
@@ -36,16 +36,16 @@ func (s *aztestsSuite) TestRetryTestScenarioUntilMaxRetries(c *chk.C) {
 	testRetryTestScenario(c, retryTestScenarioRetryUntilMaxRetries)
 }
 
-func newRetryTestPolicyFactory(c *chk.C, scenario retryTestScenario, maxRetries int, cancel context.CancelFunc) *retryTestPolicyFactory {
+func newRetryTestPolicyFactory(c *chk.C, scenario retryTestScenario, maxRetries int32, cancel context.CancelFunc) *retryTestPolicyFactory {
 	return &retryTestPolicyFactory{c: c, scenario: scenario, maxRetries: maxRetries, cancel: cancel}
 }
 
 type retryTestPolicyFactory struct {
 	c          *chk.C
 	scenario   retryTestScenario
-	maxRetries int
+	maxRetries int32
 	cancel     context.CancelFunc
-	try        int
+	try        int32
 }
 
 func (f *retryTestPolicyFactory) New(node pipeline.Node) pipeline.Policy {
@@ -60,7 +60,6 @@ type retryTestPolicy struct {
 
 type retryError struct {
 	temporary, timeout bool
-	statusCode         int
 }
 
 func (e *retryError) Temporary() bool { return e.temporary }