// GetWorkerUnsafe returns a TimedWorker corresponding to the given key.
// Unsafe method - workers have attached goroutines which can fire afater this function is called.
439486
func (q *TimedWorkerQueue) GetWorkerUnsafe(key string) *TimedWorker {
	q.Lock()
	defer q.Unlock()
	return q.workers[key]
}
747019
func (e *Enforcer) EnforceSafe(rvals ...interface{}) (result bool, err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("%v", r)
			result = false
		}
	}()

	result = e.Enforce(rvals...)
	err = nil
	return
}
512630
func (e *SyncedEnforcer) EnforceSafe(rvals ...interface{}) (result bool, err error) {
	e.m.RLock()
	defer e.m.RUnlock()
	return e.Enforcer.EnforceSafe(rvals...)
}
1212939
func GetServiceNetworkTier(service *v1.Service) (cloud.NetworkTier, error) {
	l, ok := service.Annotations[NetworkTierAnnotationKey]
	if !ok {
		return cloud.NetworkTierDefault, nil
	}

	v := cloud.NetworkTier(l)
	switch v {
	case cloud.NetworkTierStandard:
		fallthrough
	case cloud.NetworkTierPremium:
		return v, nil
	default:
		return cloud.NetworkTierDefault, fmt.Errorf("unsupported network tier: %q", v)
	}
}
102659
func (c *Client) GetTaskEventHandler(allocID, taskName string) drivermanager.EventHandler {
	c.allocLock.RLock()
	defer c.allocLock.RUnlock()
	if ar, ok := c.allocs[allocID]; ok {
		return ar.GetTaskEventHandler(taskName)
	}
	return nil
}
221168
func (ctrl *PersistentVolumeController) claimWorker() {
	workFunc := func() bool {
		keyObj, quit := ctrl.claimQueue.Get()
		if quit {
			return true
		}
		defer ctrl.claimQueue.Done(keyObj)
		key := keyObj.(string)
		klog.V(5).Infof("claimWorker[%s]", key)

		namespace, name, err := cache.SplitMetaNamespaceKey(key)
		if err != nil {
			klog.V(4).Infof("error getting namespace & name of claim %q to get claim from informer: %v", key, err)
			return false
		}
		claim, err := ctrl.claimLister.PersistentVolumeClaims(namespace).Get(name)
		if err == nil {
			// The claim still exists in informer cache, the event must have
			// been add/update/sync
			ctrl.updateClaim(claim)
			return false
		}
		if !errors.IsNotFound(err) {
			klog.V(2).Infof("error getting claim %q from informer: %v", key, err)
			return false
		}

		// The claim is not in informer cache, the event must have been "delete"
		claimObj, found, err := ctrl.claims.GetByKey(key)
		if err != nil {
			klog.V(2).Infof("error getting claim %q from cache: %v", key, err)
			return false
		}
		if !found {
			// The controller has already processed the delete event and
			// deleted the claim from its cache
			klog.V(2).Infof("deletion of claim %q was already processed", key)
			return false
		}
		claim, ok := claimObj.(*v1.PersistentVolumeClaim)
		if !ok {
			klog.Errorf("expected claim, got %+v", claimObj)
			return false
		}
		ctrl.deleteClaim(claim)
		return false
	}
	for {
		if quit := workFunc(); quit {
			klog.Infof("claim worker queue shutting down")
			return
		}
	}
}
1217198
func (conn *Conn) inWorker() {
	for {
		msg, err := conn.ReadMessage()
		if err != nil {
			if _, ok := err.(InvalidMessageError); !ok {
				// Some read error occured (usually EOF); we can't really do
				// anything but to shut down all stuff and returns errors to all
				// pending replies.
				conn.Close()
				conn.calls.finalizeAllWithError(err)
				return
			}
			// invalid messages are ignored
			continue
		}
		conn.eavesdroppedLck.Lock()
		if conn.eavesdropped != nil {
			select {
			case conn.eavesdropped <- msg:
			default:
			}
			conn.eavesdroppedLck.Unlock()
			continue
		}
		conn.eavesdroppedLck.Unlock()
		dest, _ := msg.Headers[FieldDestination].value.(string)
		found := dest == "" ||
			!conn.names.uniqueNameIsKnown() ||
			conn.names.isKnownName(dest)
		if !found {
			// Eavesdropped a message, but no channel for it is registered.
			// Ignore it.
			continue
		}
		switch msg.Type {
		case TypeError:
			conn.serialGen.RetireSerial(conn.calls.handleDBusError(msg))
		case TypeMethodReply:
			conn.serialGen.RetireSerial(conn.calls.handleReply(msg))
		case TypeSignal:
			conn.handleSignal(msg)
		case TypeMethodCall:
			go conn.handleCall(msg)
		}

	}
}
188365
func isNetworkError() hpredicate {
	return func(c *context) bool {
		return c.responseCode == http.StatusBadGateway || c.responseCode == http.StatusGatewayTimeout
	}
}
1183157
func GetKeyHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
	defer r.Body.Close()
	vars := mux.Vars(r)
	return getKeyHandler(ctx, w, r, vars)
}
572011
func GetTimeErr(key string, args ...interface{}) (time.Time, error) {
	return std.GetTimeErr(key, args...)
}
937506
func NewStaticFlagWorker(value bool) worker.Worker {
	w := &staticFlagWorker{value: value}
	w.tomb.Go(func() error {
		<-w.tomb.Dying()
		return tomb.ErrDying
	})
	return w
}
903458
func isKeySafe(s []byte) bool {
	return whitelistPattern.Match(s) && !blacklistPattern.Match(s)
}
895163
func newWaitWorker(config Config, targetVersion int) (worker.Worker, error) {
	watcher, err := config.Facade.WatchModelEnvironVersion(config.ModelTag)
	if err != nil {
		return nil, errors.Trace(err)
	}
	ww := waitWorker{
		watcher:       watcher,
		facade:        config.Facade,
		modelTag:      config.ModelTag,
		gate:          config.GateUnlocker,
		targetVersion: targetVersion,
	}
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &ww.catacomb,
		Init: []worker.Worker{watcher},
		Work: ww.loop,
	}); err != nil {
		return nil, errors.Trace(err)
	}
	return &ww, nil
}
354514
func (e *SyncedEnforcer) GetPermissionsForUser(user string) [][]string {
	e.m.Lock()
	defer e.m.Unlock()
	return e.Enforcer.GetPermissionsForUser(user)
}
573652
func (api *API) DeleteWorker(requestParams *WorkerRequestParams) (WorkerScriptResponse, error) {
	// if ScriptName is provided we will treat as org request
	if requestParams.ScriptName != "" {
		return api.deleteWorkerWithName(requestParams.ScriptName)
	}
	uri := "/zones/" + requestParams.ZoneID + "/workers/script"
	res, err := api.makeRequest("DELETE", uri, nil)
	var r WorkerScriptResponse
	if err != nil {
		return r, errors.Wrap(err, errMakeRequestError)
	}
	err = json.Unmarshal(res, &r)
	if err != nil {
		return r, errors.Wrap(err, errUnmarshalError)
	}
	return r, nil
}
13619
func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask, ch chan<- *copResponse) ([]*copTask, error) {
	failpoint.Inject("handleTaskOnceError", func(val failpoint.Value) {
		if val.(bool) {
			failpoint.Return(nil, errors.New("mock handleTaskOnce error"))
		}
	})

	sender := NewRegionRequestSender(worker.store.regionCache, worker.store.client)
	req := &tikvrpc.Request{
		Type: task.cmdType,
		Cop: &coprocessor.Request{
			Tp:     worker.req.Tp,
			Data:   worker.req.Data,
			Ranges: task.ranges.toPBRanges(),
		},
		Context: kvrpcpb.Context{
			IsolationLevel: pbIsolationLevel(worker.req.IsolationLevel),
			Priority:       kvPriorityToCommandPri(worker.req.Priority),
			NotFillCache:   worker.req.NotFillCache,
			HandleTime:     true,
			ScanDetail:     true,
		},
	}
	startTime := time.Now()
	resp, rpcCtx, err := sender.SendReqCtx(bo, req, task.region, ReadTimeoutMedium)
	if err != nil {
		return nil, errors.Trace(err)
	}
	// Set task.storeAddr field so its task.String() method have the store address information.
	task.storeAddr = sender.storeAddr
	costTime := time.Since(startTime)
	if costTime > minLogCopTaskTime {
		worker.logTimeCopTask(costTime, task, bo, resp)
	}
	metrics.TiKVCoprocessorHistogram.Observe(costTime.Seconds())

	if task.cmdType == tikvrpc.CmdCopStream {
		return worker.handleCopStreamResult(bo, rpcCtx, resp.CopStream, task, ch)
	}

	// Handles the response for non-streaming copTask.
	return worker.handleCopResponse(bo, rpcCtx, &copResponse{pbResp: resp.Cop}, task, ch, nil)
}
559424
func (o *SAMLConnectorV2) GetServiceProvider(clock clockwork.Clock) (*saml2.SAMLServiceProvider, error) {
	if o.Metadata.Name == "" {
		return nil, trace.BadParameter("ID: missing connector name, name your connector to refer to internally e.g. okta1")
	}
	if o.Metadata.Name == teleport.Local {
		return nil, trace.BadParameter("ID: invalid connector name %v is a reserved name", teleport.Local)
	}
	if o.Spec.AssertionConsumerService == "" {
		return nil, trace.BadParameter("missing acs - assertion consumer service parameter, set service URL that will receive POST requests from SAML")
	}
	if o.Spec.ServiceProviderIssuer == "" {
		o.Spec.ServiceProviderIssuer = o.Spec.AssertionConsumerService
	}
	if o.Spec.Audience == "" {
		o.Spec.Audience = o.Spec.AssertionConsumerService
	}
	certStore := dsig.MemoryX509CertificateStore{
		Roots: []*x509.Certificate{},
	}
	// if we have a entity descriptor url, fetch it first
	if o.Spec.EntityDescriptorURL != "" {
		resp, err := http.Get(o.Spec.EntityDescriptorURL)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		if resp.StatusCode != http.StatusOK {
			return nil, trace.BadParameter("status code %v when fetching from %q", resp.StatusCode, o.Spec.EntityDescriptorURL)
		}
		defer resp.Body.Close()
		body, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		o.Spec.EntityDescriptor = string(body)
		log.Debugf("[SAML] Successfully fetched entity descriptor from %q", o.Spec.EntityDescriptorURL)
	}
	if o.Spec.EntityDescriptor != "" {
		metadata := &types.EntityDescriptor{}
		err := xml.Unmarshal([]byte(o.Spec.EntityDescriptor), metadata)
		if err != nil {
			return nil, trace.Wrap(err, "failed to parse entity_descriptor")
		}

		for _, kd := range metadata.IDPSSODescriptor.KeyDescriptors {
			certData, err := base64.StdEncoding.DecodeString(strings.TrimSpace(kd.KeyInfo.X509Data.X509Certificate.Data))
			if err != nil {
				return nil, trace.Wrap(err)
			}
			cert, err := x509.ParseCertificate(certData)
			if err != nil {
				return nil, trace.Wrap(err, "failed to parse certificate in metadata")
			}
			certStore.Roots = append(certStore.Roots, cert)
		}
		o.Spec.Issuer = metadata.EntityID
		o.Spec.SSO = metadata.IDPSSODescriptor.SingleSignOnService.Location
	}
	if o.Spec.Issuer == "" {
		return nil, trace.BadParameter("no issuer or entityID set, either set issuer as a parameter or via entity_descriptor spec")
	}
	if o.Spec.SSO == "" {
		return nil, trace.BadParameter("no SSO set either explicitly or via entity_descriptor spec")
	}
	if o.Spec.Cert != "" {
		cert, err := utils.ParseCertificatePEM([]byte(o.Spec.Cert))
		if err != nil {
			return nil, trace.Wrap(err)
		}
		certStore.Roots = append(certStore.Roots, cert)
	}
	if len(certStore.Roots) == 0 {
		return nil, trace.BadParameter(
			"no identity provider certificate provided, either set certificate as a parameter or via entity_descriptor")
	}
	if o.Spec.SigningKeyPair == nil {
		keyPEM, certPEM, err := utils.GenerateSelfSignedSigningCert(pkix.Name{
			Organization: []string{"Teleport OSS"},
			CommonName:   "teleport.localhost.localdomain",
		}, nil, 10*365*24*time.Hour)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		o.Spec.SigningKeyPair = &SigningKeyPair{
			PrivateKey: string(keyPEM),
			Cert:       string(certPEM),
		}
	}
	keyStore, err := utils.ParseSigningKeyStorePEM(o.Spec.SigningKeyPair.PrivateKey, o.Spec.SigningKeyPair.Cert)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	// make sure claim mappings have either roles or a role template
	for _, v := range o.Spec.AttributesToRoles {
		hasRoles := false
		if len(v.Roles) > 0 {
			hasRoles = true
		}
		hasRoleTemplate := false
		if v.RoleTemplate != nil {
			hasRoleTemplate = true
		}

		// we either need to have roles or role templates not both or neither
		// ! ( hasRoles XOR hasRoleTemplate )
		if hasRoles == hasRoleTemplate {
			return nil, trace.BadParameter("need roles or role template (not both or none)")
		}
	}
	log.Debugf("[SAML] SSO: %v", o.Spec.SSO)
	log.Debugf("[SAML] Issuer: %v", o.Spec.Issuer)
	log.Debugf("[SAML] ACS: %v", o.Spec.AssertionConsumerService)

	sp := &saml2.SAMLServiceProvider{
		IdentityProviderSSOURL:         o.Spec.SSO,
		IdentityProviderIssuer:         o.Spec.Issuer,
		ServiceProviderIssuer:          o.Spec.ServiceProviderIssuer,
		AssertionConsumerServiceURL:    o.Spec.AssertionConsumerService,
		SignAuthnRequests:              true,
		SignAuthnRequestsCanonicalizer: dsig.MakeC14N11Canonicalizer(),
		AudienceURI:                    o.Spec.Audience,
		IDPCertificateStore:            &certStore,
		SPKeyStore:                     keyStore,
		Clock:                          dsig.NewFakeClock(clock),
		NameIdFormat:                   "urn:oasis:names:tc:SAML:1.1:nameid-format:unspecified",
	}

	// adfs specific settings
	if o.Spec.Provider == teleport.ADFS {
		if sp.SignAuthnRequests {
			// adfs does not support C14N11, we have to use the C14N10 canonicalizer
			sp.SignAuthnRequestsCanonicalizer = dsig.MakeC14N10ExclusiveCanonicalizerWithPrefixList(dsig.DefaultPrefix)

			// at a minimum we require password protected transport
			sp.RequestedAuthnContext = &saml2.RequestedAuthnContext{
				Comparison: "minimum",
				Contexts:   []string{"urn:oasis:names:tc:SAML:2.0:ac:classes:PasswordProtectedTransport"},
			}
		}
	}

	return sp, nil
}