<commit_before>/*
Copyright 2018 Google, Inc. All rights reserved.

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 fetcher

import (
	"archive/zip"
	"context"
	"crypto/sha1"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"math"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/GoogleCloudPlatform/cloud-builders/gcs-fetcher/pkg/common"
)

var (
	// sourceExt is a best-effort to identify files that should have a short
	// download time and thus get a short timeout for the first tries.
	sourceExt = map[string]bool{
		".js":   true,
		".py":   true,
		".php":  true,
		".java": true,
		".go":   true,
		".cs":   true,
		".rb":   true,
		".css":  true,
		".vb":   true,
		".pl":   true, // Perl. C'mon, live a little.
	}
	sourceTimeout = map[int]time.Duration{ // try number -> timeout duration
		0: 1 * time.Second,
		1: 2 * time.Second,
	}
	notSourceTimeout = map[int]time.Duration{ // try number -> timeout duration
		0: 3 * time.Second,
		1: 6 * time.Second,
	}
	defaultTimeout = 1 * time.Hour
	noTimeout      = 0 * time.Second
	errGCSTimeout  = errors.New("GCS timeout")
)

type sizeBytes int64

// job is a file to download, corresponds to an entry in the manifest file.
type job struct {
	filename       string
	bucket, object string
	generation     int64
	sha1sum        string
}

// jobAttempt is an attempt to download a particular file, may result in
// success or failure (indicated by err).
type jobAttempt struct {
	started    time.Time
	duration   time.Duration
	err        error
	gcsTimeout time.Duration
}

// jobReport stores all the details about the attempts to download a
// particular file.
type jobReport struct {
	job       job
	started   time.Time
	completed time.Time
	size      sizeBytes
	attempts  []jobAttempt
	success   bool
	finalname string
	err       error
}

type fetchOnceResult struct {
	size sizeBytes
	err  error
}

type stats struct {
	workers     int
	files       int
	size        sizeBytes
	duration    time.Duration
	retries     int
	gcsTimeouts int
	success     bool
	errs        []error
}

// OS allows us to inject dependencies to facilitate testing.
type OS interface {
	Rename(oldpath, newpath string) error
	Chmod(name string, mode os.FileMode) error
	Create(name string) (*os.File, error)
	MkdirAll(path string, perm os.FileMode) error
	Open(name string) (*os.File, error)
	RemoveAll(path string) error
}

// GCS allows us to inject dependencies to facilitate testing.
type GCS interface {
	NewReader(ctx context.Context, bucket, object string) (io.ReadCloser, error)
}

// Fetcher is the main workhorse of this package and does all the heavy lifting.
type Fetcher struct {
	GCS GCS
	OS  OS

	DestDir    string
	StagingDir string

	// mu guards CreatedDirs
	mu          sync.Mutex
	CreatedDirs map[string]bool

	SourceType     string
	Bucket, Object string
	Generation     int64

	TimeoutGCS  bool
	WorkerCount int
	Retries     int
	Backoff     time.Duration
	Verbose     bool
}

func (gf *Fetcher) recordFailure(j job, started time.Time, gcsTimeout time.Duration, err error, report *jobReport) {
	attempt := jobAttempt{
		started:    started,
		duration:   time.Since(started),
		err:        err,
		gcsTimeout: gcsTimeout,
	}
	report.success = false
	report.err = err // Hold the latest error.
	report.attempts = append(report.attempts, attempt)

	isLast := len(report.attempts) == gf.Retries
	if gf.Verbose || isLast {
		retryMsg := ", will retry"
		if isLast {
			retryMsg = ", will no longer retry"
		}
		log.Printf("Failed to fetch %s%s: %v", formatGCSName(j.bucket, j.object, j.generation), retryMsg, err)
	}
}

func (gf *Fetcher) recordSuccess(j job, started time.Time, size sizeBytes, finalname string, report *jobReport) {
	attempt := jobAttempt{
		started:  started,
		duration: time.Since(started),
	}
	report.success = true
	report.err = nil
	report.size = size
	report.attempts = append(report.attempts, attempt)
	report.finalname = finalname

	mibps := math.MaxFloat64
	if attempt.duration > 0 {
		mibps = (float64(report.size) / 1024 / 1024) / attempt.duration.Seconds()
	}
	log.Printf("Fetched %s (%dB in %v, %.2fMiB/s)", formatGCSName(j.bucket, j.object, j.generation), report.size, attempt.duration, mibps)
}

// fetchObject is responsible for trying (and retrying) to fetch a single file
// from GCS. It first downloads the file to a temp file, then renames it to
// the final location and sets the permissions on the final file.
func (gf *Fetcher) fetchObject(ctx context.Context, j job) *jobReport {
	report := &jobReport{job: j, started: time.Now()}
	defer func() {
		report.completed = time.Now()
	}()

	var tmpfile string
	var backoff time.Duration

	for retrynum := 0; retrynum <= gf.Retries; retrynum++ {
		// Apply appropriate retry backoff.
		if retrynum > 0 {
			if retrynum == 1 {
				backoff = gf.Backoff
			} else {
				backoff *= 2<commit_msg>Output frequent log line only if --verbose=true. (#370)<commit_after>/*
Copyright 2018 Google, Inc. All rights reserved.

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 fetcher

import (
	"archive/zip"
	"context"
	"crypto/sha1"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"math"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/GoogleCloudPlatform/cloud-builders/gcs-fetcher/pkg/common"
)

var (
	// sourceExt is a best-effort to identify files that should have a short
	// download time and thus get a short timeout for the first tries.
	sourceExt = map[string]bool{
		".js":   true,
		".py":   true,
		".php":  true,
		".java": true,
		".go":   true,
		".cs":   true,
		".rb":   true,
		".css":  true,
		".vb":   true,
		".pl":   true, // Perl. C'mon, live a little.
	}
	sourceTimeout = map[int]time.Duration{ // try number -> timeout duration
		0: 1 * time.Second,
		1: 2 * time.Second,
	}
	notSourceTimeout = map[int]time.Duration{ // try number -> timeout duration
		0: 3 * time.Second,
		1: 6 * time.Second,
	}
	defaultTimeout = 1 * time.Hour
	noTimeout      = 0 * time.Second
	errGCSTimeout  = errors.New("GCS timeout")
)

type sizeBytes int64

// job is a file to download, corresponds to an entry in the manifest file.
type job struct {
	filename       string
	bucket, object string
	generation     int64
	sha1sum        string
}

// jobAttempt is an attempt to download a particular file, may result in
// success or failure (indicated by err).
type jobAttempt struct {
	started    time.Time
	duration   time.Duration
	err        error
	gcsTimeout time.Duration
}

// jobReport stores all the details about the attempts to download a
// particular file.
type jobReport struct {
	job       job
	started   time.Time
	completed time.Time
	size      sizeBytes
	attempts  []jobAttempt
	success   bool
	finalname string
	err       error
}

type fetchOnceResult struct {
	size sizeBytes
	err  error
}

type stats struct {
	workers     int
	files       int
	size        sizeBytes
	duration    time.Duration
	retries     int
	gcsTimeouts int
	success     bool
	errs        []error
}

// OS allows us to inject dependencies to facilitate testing.
type OS interface {
	Rename(oldpath, newpath string) error
	Chmod(name string, mode os.FileMode) error
	Create(name string) (*os.File, error)
	MkdirAll(path string, perm os.FileMode) error
	Open(name string) (*os.File, error)
	RemoveAll(path string) error
}

// GCS allows us to inject dependencies to facilitate testing.
type GCS interface {
	NewReader(ctx context.Context, bucket, object string) (io.ReadCloser, error)
}

// Fetcher is the main workhorse of this package and does all the heavy lifting.
type Fetcher struct {
	GCS GCS
	OS  OS

	DestDir    string
	StagingDir string

	// mu guards CreatedDirs
	mu          sync.Mutex
	CreatedDirs map[string]bool

	SourceType     string
	Bucket, Object string
	Generation     int64

	TimeoutGCS  bool
	WorkerCount int
	Retries     int
	Backoff     time.Duration
	Verbose     bool
}

func (gf *Fetcher) recordFailure(j job, started time.Time, gcsTimeout time.Duration, err error, report *jobReport) {
	attempt := jobAttempt{
		started:    started,
		duration:   time.Since(started),
		err:        err,
		gcsTimeout: gcsTimeout,
	}
	report.success = false
	report.err = err // Hold the latest error.
	report.attempts = append(report.attempts, attempt)

	isLast := len(report.attempts) == gf.Retries
	if gf.Verbose || isLast {
		retryMsg := ", will retry"
		if isLast {
			retryMsg = ", will no longer retry"
		}
		log.Printf("Failed to fetch %s%s: %v", formatGCSName(j.bucket, j.object, j.generation), retryMsg, err)
	}
}

func (gf *Fetcher) recordSuccess(j job, started time.Time, size sizeBytes, finalname string, report *jobReport) {
	attempt := jobAttempt{
		started:  started,
		duration: time.Since(started),
	}
	report.success = true
	report.err = nil
	report.size = size
	report.attempts = append(report.attempts, attempt)
	report.finalname = finalname

	mibps := math.MaxFloat64
	if attempt.duration > 0 {
		mibps = (float64(report.size) / 1024 / 1024) / attempt.duration.Seconds()
	}
	if gf.Verbose {
	  log.Printf("Fetched %s (%dB in %v, %.2fMiB/s)", formatGCSName(j.bucket, j.object, j.generation), report.size, attempt.duration, mibps)
	}
}

// fetchObject is responsible for trying (and retrying) to fetch a single file
// from GCS. It first downloads the file to a temp file, then renames it to
// the final location and sets the permissions on the final file.
func (gf *Fetcher) fetchObject(ctx context.Context, j job) *jobReport {
	report := &jobReport{job: j, started: time.Now()}
	defer func() {
		report.completed = time.Now()
	}()

	var tmpfile string
	var backoff time.Duration

	for retrynum := 0; retrynum <= gf.Retries; retrynum++ {
		// Apply appropriate retry backoff.
		if retrynum > 0 {
			if retrynum == 1 {
				backoff = gf.Backoff
			} else {
				backoff *= 2
{'fraction_non_alphanumeric': 0.09925233644859813, 'fraction_numerical': 0.008504672897196261, 'mean_word_length': 3.3253839935327405, 'pattern_counts': {'":': 20, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '345630', 'n_tokens_mistral': 3671, 'n_tokens_neox': 3441, 'n_words': 1214}