// runMirror - mirrors all buckets to another S3 server
437378
func runMirror(srcURL, dstURL string, ctx *cli.Context, encKeyDB map[string][]prefixSSEPair) bool {
	// This is kept for backward compatibility, `--force` means
	// --overwrite.
	isOverwrite := ctx.Bool("force")
	if !isOverwrite {
		isOverwrite = ctx.Bool("overwrite")
	}

	// Create a new mirror job and execute it
	mj := newMirrorJob(srcURL, dstURL,
		ctx.Bool("fake"),
		ctx.Bool("remove"),
		isOverwrite,
		ctx.Bool("watch"),
		ctx.StringSlice("exclude"),
		ctx.String("older-than"),
		ctx.String("newer-than"),
		ctx.String("storage-class"),
		encKeyDB)

	srcClt, err := newClient(srcURL)
	fatalIf(err, "Unable to initialize `"+srcURL+"`.")

	dstClt, err := newClient(dstURL)
	fatalIf(err, "Unable to initialize `"+srcURL+"`.")

	if ctx.Bool("a") && (srcClt.GetURL().Type != objectStorage || dstClt.GetURL().Type != objectStorage) {
		fatalIf(errDummy(), "Synchronizing bucket policies is only possible when both source & target point to S3 servers.")
	}

	mirrorAllBuckets := (srcClt.GetURL().Type == objectStorage && srcClt.GetURL().Path == "/") ||
		(dstClt.GetURL().Type == objectStorage && dstClt.GetURL().Path == "/")

	if mirrorAllBuckets {
		// Synchronize buckets using dirDifference function
		for d := range dirDifference(srcClt, dstClt, srcURL, dstURL) {
			if d.Error != nil {
				mj.status.fatalIf(d.Error, fmt.Sprintf("Failed to start monitoring."))
			}
			if d.Diff == differInSecond {
				// Ignore buckets that only exist in target instance
				continue
			}

			sourceSuffix := strings.TrimPrefix(d.FirstURL, srcClt.GetURL().String())
			newSrcURL := path.Join(srcURL, sourceSuffix)
			newTgtURL := path.Join(dstURL, sourceSuffix)

			newSrcClt, _ := newClient(newSrcURL)
			newDstClt, _ := newClient(newTgtURL)

			if d.Diff == differInFirst {
				// Bucket only exists in the source, create the same bucket in the destination
				if err := newDstClt.MakeBucket(ctx.String("region"), false); err != nil {
					errorIf(err, "Cannot created bucket in `"+newTgtURL+"`.")
					continue
				}
				// Copy policy rules from source to dest if flag is activated
				if ctx.Bool("a") {
					if err := copyBucketPolicies(srcClt, dstClt, isOverwrite); err != nil {
						errorIf(err, "Cannot copy bucket policies to `"+newDstClt.GetURL().String()+"`.")
					}
				}
			}

			if mj.isWatch {
				// monitor mode will watch the source folders for changes,
				// and queue them for copying.
				if err := mj.watchURL(newSrcClt); err != nil {
					mj.status.fatalIf(err, fmt.Sprintf("Failed to start monitoring."))
				}
			}
		}
	}

	if !mirrorAllBuckets && mj.isWatch {
		// monitor mode will watch the source folders for changes,
		// and queue them for copying.
		if err := mj.watchURL(srcClt); err != nil {
			mj.status.fatalIf(err, fmt.Sprintf("Failed to start monitoring."))
		}
	}

	ctxt, cancelMirror := context.WithCancel(context.Background())
	defer cancelMirror()

	// Start mirroring job
	return mj.mirror(ctxt, cancelMirror)
}
691521
func (a *application) runCollector(c context.Context) error {
	cfg := a.ServiceConfig()
	ccfg := cfg.GetCollector()
	if ccfg == nil {
		return errors.New("no collector configuration")
	}

	pscfg := cfg.GetTransport().GetPubsub()
	if pscfg == nil {
		return errors.New("missing Pub/Sub configuration")
	}

	// Our Subscription must be a valid one.
	sub := gcps.NewSubscription(pscfg.Project, pscfg.Subscription)
	if err := sub.Validate(); err != nil {
		return fmt.Errorf("invalid Pub/Sub subscription %q: %v", sub, err)
	}

	// New PubSub instance with the authenticated client.
	psClient, err := a.Service.PubSubSubscriberClient(c, pscfg.Project)
	if err != nil {
		log.Fields{
			log.ErrorKey:   err,
			"subscription": sub,
		}.Errorf(c, "Failed to create Pub/Sub client.")
		return err
	}

	psSub := psClient.Subscription(pscfg.Subscription)
	exists, err := psSub.Exists(c)
	if err != nil {
		log.Fields{
			log.ErrorKey:   err,
			"subscription": sub,
		}.Errorf(c, "Could not confirm Pub/Sub subscription.")
		return errInvalidConfig
	}
	psSub.ReceiveSettings = pubsub.ReceiveSettings{
		MaxExtension:           24 * time.Hour,
		MaxOutstandingMessages: int(ccfg.MaxConcurrentMessages), // If < 1, default.
		MaxOutstandingBytes:    0,                               // Default.
	}

	if !exists {
		log.Fields{
			"subscription": sub,
		}.Errorf(c, "Subscription does not exist.")
		return errInvalidConfig
	}
	log.Fields{
		"subscription": sub,
	}.Infof(c, "Successfully validated Pub/Sub subscription.")

	st, err := a.IntermediateStorage(c, true)
	if err != nil {
		return err
	}
	defer st.Close()

	// Initialize a Coordinator client that bundles requests together.
	coordClient := &bundleServicesClient.Client{
		ServicesClient:       a.Coordinator(),
		DelayThreshold:       time.Second,
		BundleCountThreshold: 100,
	}
	defer coordClient.Flush()

	// Initialize our Collector service object using a caching Coordinator
	// interface.
	coord := coordinator.NewCoordinator(coordClient)
	coord = coordinator.NewCache(coord, int(ccfg.StateCacheSize), google.DurationFromProto(ccfg.StateCacheExpiration))

	coll := collector.Collector{
		Coordinator:       coord,
		Storage:           st,
		MaxMessageWorkers: int(ccfg.MaxMessageWorkers),
	}
	defer coll.Close()

	c, cancelFunc := context.WithCancel(c)
	defer cancelFunc()

	// Application shutdown will now operate by cancelling the Collector's
	// shutdown Context.
	a.SetShutdownFunc(cancelFunc)

	// Execute our main subscription pull loop. It will run until the supplied
	// Context is cancelled.
	retryForever := func() retry.Iterator {
		return &retry.ExponentialBackoff{
			Limited: retry.Limited{
				Delay:   200 * time.Millisecond,
				Retries: -1, // Unlimited.
			},
			MaxDelay:   10 * time.Second,
			Multiplier: 2,
		}
	}

	err = retry.Retry(c, transient.Only(retryForever), func() error {
		return grpcutil.WrapIfTransient(psSub.Receive(c, func(c context.Context, msg *pubsub.Message) {
			c = log.SetField(c, "messageID", msg.ID)
			if a.processMessage(c, &coll, msg) {
				// ACK the message, removing it from Pub/Sub.
				msg.Ack()
			} else {
				// NACK the message. It will be redelivered and processed.
				msg.Nack()
			}
		}))
	}, func(err error, d time.Duration) {
		log.Fields{
			log.ErrorKey: err,
			"delay":      d,
		}.Warningf(c, "Transient error during subscription Receive loop; retrying...")
	})
	if err != nil {
		log.WithError(err).Errorf(c, "Failed during Pub/Sub Receive.")
		return err
	}

	log.Debugf(c, "Collector finished.")
	return nil
}
1315770
func (hk *HollowKubelet) Run() {
	if err := kubeletapp.RunKubelet(&options.KubeletServer{
		KubeletFlags:         *hk.KubeletFlags,
		KubeletConfiguration: *hk.KubeletConfiguration,
	}, hk.KubeletDeps, false); err != nil {
		klog.Fatalf("Failed to run HollowKubelet: %v. Exiting.", err)
	}
	select {}
}
1059681
func (s *Bucket) PutExternalMirrorRequest(input *PutBucketExternalMirrorInput) (*request.Request, *PutBucketExternalMirrorOutput, error) {

	if input == nil {
		input = &PutBucketExternalMirrorInput{}
	}

	properties := *s.Properties

	o := &data.Operation{
		Config:        s.Config,
		Properties:    &properties,
		APIName:       "PUT Bucket External Mirror",
		RequestMethod: "PUT",
		RequestURI:    "/<bucket-name>?mirror",
		StatusCodes: []int{
			200, // OK
		},
	}

	x := &PutBucketExternalMirrorOutput{}
	r, err := request.New(o, input, x)
	if err != nil {
		return nil, nil, err
	}

	return r, x, nil
}
344001
func overlayRender(cfg RunConfig, treeStoreID string, cdir string, dest string, appName string) error {
	cachedTreePath := cfg.TreeStore.GetRootFS(treeStoreID)
	mc, err := prepareOverlay(cachedTreePath, treeStoreID, cdir, dest, appName, cfg.MountLabel,
		cfg.RktGid, common.DefaultRegularDirPerm)
	if err != nil {
		return errwrap.Wrap(errors.New("problem preparing overlay directories"), err)
	}
	if err = overlay.Mount(mc); err != nil {
		return errwrap.Wrap(errors.New("problem mounting overlay filesystem"), err)
	}

	return nil
}
746308
func MirrorUpdate() {
	if taskStatusTable.IsRunning(_MIRROR_UPDATE) {
		return
	}
	taskStatusTable.Start(_MIRROR_UPDATE)
	defer taskStatusTable.Stop(_MIRROR_UPDATE)

	log.Trace("Doing: MirrorUpdate")

	if err := x.Where("next_update_unix<=?", time.Now().Unix()).Iterate(new(Mirror), func(idx int, bean interface{}) error {
		m := bean.(*Mirror)
		if m.Repo == nil {
			log.Error(2, "Disconnected mirror repository found: %d", m.ID)
			return nil
		}

		MirrorQueue.Add(m.RepoID)
		return nil
	}); err != nil {
		log.Error(2, "MirrorUpdate: %v", err)
	}
}
980116
func (e *MirrorSecrets) Run(c *fi.Context) error {
	return fi.DefaultDeltaRunMethod(e, c)
}
665955
func (s *Bucket) DeleteExternalMirror() (*DeleteBucketExternalMirrorOutput, error) {
	r, x, err := s.DeleteExternalMirrorRequest()

	if err != nil {
		return x, err
	}

	err = r.Send()
	if err != nil {
		return nil, err
	}

	requestID := r.HTTPResponse.Header.Get(http.CanonicalHeaderKey("X-QS-Request-ID"))
	x.RequestID = &requestID

	return x, err
}
1131198
func (mj *mirrorJob) watchMirror(ctx context.Context, cancelMirror context.CancelFunc) {
	for {
		select {
		case event, ok := <-mj.watcher.Events():
			if !ok {
				return
			}

			// It will change the expanded alias back to the alias
			// again, by replacing the sourceUrlFull with the sourceAlias.
			// This url will be used to mirror.
			sourceAlias, sourceURLFull, _ := mustExpandAlias(mj.sourceURL)

			// If the passed source URL points to fs, fetch the absolute src path
			// to correctly calculate targetPath
			if sourceAlias == "" {
				tmpSrcURL, err := filepath.Abs(sourceURLFull)
				if err == nil {
					sourceURLFull = tmpSrcURL
				}
			}
			eventPath := event.Path

			if runtime.GOOS == "darwin" {
				// Strip the prefixes in the event path. Happens in darwin OS only
				eventPath = eventPath[strings.Index(eventPath, sourceURLFull):len(eventPath)]
			}

			sourceURL := newClientURL(eventPath)
			// trim trailing slash from source url
			sourceURLStr := strings.TrimSuffix(sourceURLFull, string(sourceURL.Separator))
			aliasedPath := strings.Replace(eventPath, sourceURLStr, mj.sourceURL, -1)

			// build target path, it is the relative of the eventPath with the sourceUrl
			// joined to the targetURL.
			sourceSuffix := strings.TrimPrefix(eventPath, sourceURLFull)
			//Skip the object, if it matches the Exclude options provided
			if matchExcludeOptions(mj.excludeOptions, sourceSuffix) {
				continue
			}

			targetPath := urlJoinPath(mj.targetURL, sourceSuffix)

			// newClient needs the unexpanded  path, newCLientURL needs the expanded path
			targetAlias, expandedTargetPath, _ := mustExpandAlias(targetPath)
			targetURL := newClientURL(expandedTargetPath)
			sourcePath := filepath.ToSlash(filepath.Join(sourceAlias, sourceURL.Path))
			srcSSE := getSSE(sourcePath, mj.encKeyDB[sourceAlias])
			tgtSSE := getSSE(targetPath, mj.encKeyDB[targetAlias])

			if event.Type == EventCreate {
				// we are checking if a destination file exists now, and if we only
				// overwrite it when force is enabled.
				mirrorURL := URLs{
					SourceAlias:   sourceAlias,
					SourceContent: &clientContent{URL: *sourceURL},
					TargetAlias:   targetAlias,
					TargetContent: &clientContent{URL: *targetURL},
					encKeyDB:      mj.encKeyDB,
				}
				if event.Size == 0 {
					sourceClient, err := newClient(aliasedPath)
					if err != nil {
						// cannot create sourceclient
						mj.statusCh <- mirrorURL.WithError(err)
						continue
					}
					sourceContent, err := sourceClient.Stat(false, false, srcSSE)
					if err != nil {
						// source doesn't exist anymore
						mj.statusCh <- mirrorURL.WithError(err)
						continue
					}
					targetClient, err := newClient(targetPath)
					if err != nil {
						// cannot create targetclient
						mj.statusCh <- mirrorURL.WithError(err)
						return
					}
					shouldQueue := false
					if !mj.isOverwrite {
						_, err = targetClient.Stat(false, false, tgtSSE)
						if err == nil {
							continue
						} // doesn't exist
						shouldQueue = true
					}
					if shouldQueue || mj.isOverwrite {
						mirrorURL.TotalCount = mj.TotalObjects
						mirrorURL.TotalSize = mj.TotalBytes
						// adjust total, because we want to show progress of the item still queued to be copied.
						mj.status.SetTotal(mj.status.Total() + sourceContent.Size).Update()
						mj.statusCh <- mj.doMirror(ctx, cancelMirror, mirrorURL)
					}
					continue
				}
				shouldQueue := false
				if !mj.isOverwrite {
					targetClient, err := newClient(targetPath)
					if err != nil {
						// cannot create targetclient
						mj.statusCh <- mirrorURL.WithError(err)
						return
					}
					_, err = targetClient.Stat(false, false, tgtSSE)
					if err == nil {
						continue
					} // doesn't exist
					shouldQueue = true
				}
				if shouldQueue || mj.isOverwrite {
					mirrorURL.SourceContent.Size = event.Size
					mirrorURL.TotalCount = mj.TotalObjects
					mirrorURL.TotalSize = mj.TotalBytes
					// adjust total, because we want to show progress of the itemj stiil queued to be copied.
					mj.status.SetTotal(mj.status.Total() + event.Size).Update()
					mj.statusCh <- mj.doMirror(ctx, cancelMirror, mirrorURL)
				}
			} else if event.Type == EventRemove {
				mirrorURL := URLs{
					SourceAlias:   sourceAlias,
					SourceContent: nil,
					TargetAlias:   targetAlias,
					TargetContent: &clientContent{URL: *targetURL},
					encKeyDB:      mj.encKeyDB,
				}
				mirrorURL.TotalCount = mj.TotalObjects
				mirrorURL.TotalSize = mj.TotalBytes
				if mirrorURL.TargetContent != nil && mj.isRemove {
					mj.statusCh <- mj.doRemove(mirrorURL)
				}
			}

		case err := <-mj.watcher.Errors():
			switch err.ToGoError().(type) {
			case APINotImplemented:
				errorIf(err.Trace(), "Unable to Watch on source, ignoring.")
				return
			}
			mj.statusCh <- URLs{Error: err}
			return
		case <-mj.trapCh:
			return
		}
	}
}
302541
func (mirrorRefs *MirrorRefs) getMirrorReference(datatype, contentId string, cloud CloudSpec) (*MirrorReference, error) {
	candidates := mirrorRefs.extractMirrorRefs(contentId)
	if len(candidates) == 0 {
		return nil, errors.NotFoundf("mirror data for %q", contentId)
	}
	// Restrict by cloud spec and datatype.
	hasRightCloud := func(mirrorRef *MirrorReference) bool {
		return mirrorRef.hasCloud(cloud) && mirrorRef.DataType == datatype
	}
	matchingCandidates := candidates.filter(hasRightCloud)
	if len(matchingCandidates) == 0 {
		// No cloud specific mirrors found so look for a non cloud specific mirror.
		for _, candidate := range candidates {
			if len(candidate.Clouds) == 0 {
				logger.Debugf("using default candidate for content id %q are %v", contentId, candidate)
				return &candidate, nil
			}
		}
		return nil, errors.NotFoundf("index file with cloud %v", cloud)
	}

	logger.Debugf("candidate matches for content id %q are %v", contentId, candidates)

	// Pick arbitrary match.
	return &matchingCandidates[0], nil
}
566336
func RunServer(srv *http.Server, shutdownFuncs ...func()) {
	done := make(chan struct{})

	go func() {
		// Handle SIGINT and SIGTERM.
		sigCh := make(chan os.Signal, 1)
		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

		sig := <-sigCh
		fmt.Println("Received signal, stopping.", "signal", sig)

		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		// We received an interrupt signal, shut down.
		if err := srv.Shutdown(ctx); err != nil {
			// Error from closing listeners, or context timeout:
			fmt.Printf("HTTP server Shutdown: %v\n", err)
		}
		// Run shutdown functions
		for _, sf := range shutdownFuncs {
			sf()
		}

		close(done)
	}()

	fmt.Printf("HTTP server listening on address: \"%s\"\n", srv.Addr)
	if err := srv.ListenAndServe(); err != http.ErrServerClosed {
		// Error starting or closing listener:
		fmt.Printf("HTTP server ListenAndServe: %v\n", err)
		os.Exit(1)
	}

	<-done
}
831729
func (runner *gleamRunner) runMapperReducer() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	if runner.Option.IsProfiling {
		cpuProfFile := fmt.Sprintf("mr_cpu_%d-s%d-t%d.pprof", runner.Option.HashCode,
			runner.Option.StepId, runner.Option.TaskId)
		pwd, _ := os.Getwd()
		// println("saving pprof to", pwd+"/"+cpuProfFile)

		cf, err := os.Create(cpuProfFile)
		if err != nil {
			log.Fatalf("failed to create cpu profile file %s: %v", pwd+"/"+cpuProfFile, err)
		}
		pprof.StartCPUProfile(cf)
		defer pprof.StopCPUProfile()

		memProfFile := fmt.Sprintf("mr_mem_%d-s%d-t%d.pprof", runner.Option.HashCode,
			runner.Option.StepId, runner.Option.TaskId)

		mf, err := os.Create(memProfFile)
		// println("saving pprof to", pwd+"/"+memProfFile)
		if err != nil {
			log.Fatalf("failed to create memory profile file %s: %v", pwd+"/"+memProfFile, err)
		}

		defer func() {
			runtime.GC()
			pprof.Lookup("heap").WriteTo(mf, 0)

		}()
	}

	stat.FlowHashCode = uint32(runner.Option.HashCode)
	stat.Stats = []*pb.InstructionStat{
		{
			StepId: int32(runner.Option.StepId),
			TaskId: int32(runner.Option.TaskId),
		},
	}

	if runner.Option.Mapper != "" {
		if fn, ok := mappers[MapperId(runner.Option.Mapper)]; ok {
			if err := runner.processMapper(ctx, fn.Mapper); err != nil {
				log.Fatalf("Failed to execute mapper %v: %v", os.Args, err)
			}
			return
		}
		log.Fatalf("Missing mapper function %v. Args: %v", runner.Option.Mapper, os.Args)
	}

	if runner.Option.Reducer != "" {
		if runner.Option.KeyFields == "" {
			log.Fatalf("Also expecting values for -gleam.keyFields! Actual arguments: %v", os.Args)
		}
		if fn, ok := reducers[ReducerId(runner.Option.Reducer)]; ok {
			keyPositions := strings.Split(runner.Option.KeyFields, ",")
			var keyIndexes []int
			for _, keyPosition := range keyPositions {
				keyIndex, keyIndexError := strconv.Atoi(keyPosition)
				if keyIndexError != nil {
					log.Fatalf("Failed to parse key index positions %v: %v", runner.Option.KeyFields, keyIndexError)
				}
				keyIndexes = append(keyIndexes, keyIndex)
			}

			if err := runner.processReducer(ctx, fn.Reducer, keyIndexes); err != nil {
				log.Fatalf("Failed to execute reducer %v: %v", os.Args, err)
			}
			return
		}
		log.Fatalf("Missing reducer function %v. Args: %v", runner.Option.Reducer, os.Args)

	}

	log.Fatalf("Failed to find function to execute. Args: %v", os.Args)
}
166026
func (o *RsyncOptions) RunRsync() error {
	if err := o.Strategy.Copy(o.Source, o.Destination, o.Out, o.ErrOut); err != nil {
		return err
	}

	if !o.Watch {
		return nil
	}
	return o.WatchAndSync()
}
883701
func RunAsyncRunner(ctx context.Context, tasksrv string, tasks chan task.Request, rnr *Runner) {
	u := tasksrvURL(tasksrv)

	startAsyncRunners(ctx, u, tasks, rnr)
	<-ctx.Done()
}
602064
func continueOnError(err error, mirrorEndpoint bool) bool {
	switch v := err.(type) {
	case errcode.Errors:
		if len(v) == 0 {
			return true
		}
		return continueOnError(v[0], mirrorEndpoint)
	case ErrNoSupport:
		return continueOnError(v.Err, mirrorEndpoint)
	case errcode.Error:
		return mirrorEndpoint || shouldV2Fallback(v)
	case *client.UnexpectedHTTPResponseError:
		return true
	case ImageConfigPullError:
		// ImageConfigPullError only happens with v2 images, v1 fallback is
		// unnecessary.
		// Failures from a mirror endpoint should result in fallback to the
		// canonical repo.
		return mirrorEndpoint
	case error:
		return !strings.Contains(err.Error(), strings.ToLower(syscall.ESRCH.Error()))
	}
	// let's be nice and fallback if the error is a completely
	// unexpected one.
	// If new errors have to be handled in some way, please
	// add them to the switch above.
	return true
}
357026
func HandleMirrorCredentials(url string, mosaics bool) string {
	i := strings.Index(url, "@")
	if i == -1 {
		return url
	}
	start := strings.Index(url, "://")
	if start == -1 {
		return url
	}
	if mosaics {
		return url[:start+3] + "<credentials>" + url[i:]
	}
	return url[:start+3] + url[i+1:]
}
609709
func (r *Repository) GetMirrorURL() string {
	if r == nil || r.MirrorURL == nil {
		return ""
	}
	return *r.MirrorURL
}