// EMRCluster always has an instance role of either master, core, or task
// Name is optional for core and master(only group allowed for this type) but needed for task
// since you can have multiple task instance groups.
431797
func resourceAwsEMRClusterInstanceGroupHash(v interface{}) int {
	var buf bytes.Buffer
	m := v.(map[string]interface{})
	buf.WriteString(fmt.Sprintf("%s-", m["instance_role"].(string)))
	buf.WriteString(fmt.Sprintf("%s-", m["instance_type"].(string)))
	buf.WriteString(fmt.Sprintf("%d-", m["instance_count"].(int)))
	if v, ok := m["name"]; ok {
		buf.WriteString(fmt.Sprintf("%s-", v.(string)))
	}
	if v, ok := m["bid_price"]; ok {
		buf.WriteString(fmt.Sprintf("%s-", v.(string)))
	}

	if v, ok := m["autoscaling_policy"]; ok {
		pleaseWork, _ := structure.NormalizeJsonString(v.(string))
		buf.WriteString(fmt.Sprintf("%s-", pleaseWork))
	}

	if v, ok := m["ebs_config"]; ok {
		configs := v.(*schema.Set).List()

		// There is an issue where an `ebs_config` is automatically configured when not specified in Terraform and
		// this causes the hashcode to change. Instead, we'll ignore that configuration when setting up the hashcode.
		if len(configs) > 1 {
			for _, ebsConfigs := range configs {
				buf.WriteString(fmt.Sprintf("%d-", resourceAwsEMRClusterEBSConfigHash(ebsConfigs.(map[string]interface{}))))
			}
		}
	}

	return hashcode.String(buf.String())
}
567331
func NewTerraformProvider(name, os, arch string) *TerraformProvider {
	if os == "" {
		os = runtime.GOOS
	}
	if arch == "" {
		arch = runtime.GOARCH
	}

	// separate namespace if included
	namespace := DefaultProviderNamespace
	if names := strings.SplitN(name, "/", 2); len(names) == 2 {
		namespace, name = names[0], names[1]
	}
	p := &TerraformProvider{
		RawHost:      PublicRegistryHost,
		RawNamespace: namespace,
		RawName:      name,
		OS:           os,
		Arch:         arch,
	}

	return p
}
1101941
func (a AuthorizerAdapter) GetClusterRole(name string) (*rbacv1.ClusterRole, error) {
	return a.Registry.GetClusterRole(genericapirequest.NewContext(), name, &metav1.GetOptions{})
}
1289942
func (fcd *DistributedPlanner) RunFlowContext(ctx context.Context, fc *flow.Flow) {

	stepGroups, taskGroups := plan.GroupTasks(fc)

	fmt.Println("=== task execution groups ===")
	for _, taskGroup := range taskGroups {
		fmt.Printf("%s\n", taskGroup.String())
		firstTask := taskGroup.Tasks[0]
		lastTask := taskGroup.Tasks[len(taskGroup.Tasks)-1]
		if len(firstTask.Step.InputDatasets) > 0 {
			fmt.Printf("  input:\n")
			for _, shard := range firstTask.InputShards {
				fmt.Printf("    shard: %v\n", shard.Name())
			}
		}
		if lastTask.Step.OutputDataset != nil {
			fmt.Printf("  output:\n")
			for _, shard := range lastTask.OutputShards {
				fmt.Printf("    shard: %v\n", shard.Name())
			}
		}
	}

	fmt.Println("=== step groups ===")
	for i, stepGroup := range stepGroups {
		fmt.Printf("  step group: %d", i)
		if len(stepGroup.Steps) > 0 && stepGroup.Steps[0].OutputDataset != nil {
			fmt.Printf(" partition: %d", len(stepGroup.Steps[0].OutputDataset.Shards))
		}
		fmt.Println()
		for _, step := range stepGroup.Steps {
			fmt.Printf("    step: %s", step.Name)
			if step.OutputDataset != nil {
				fmt.Printf(" size: %d MB", step.OutputDataset.GetTotalSize())
			}
			fmt.Println()
		}
	}

}
938742
func (p *Provider) DialMemContext(
	ctx context.Context,
	network string,
	laddr, raddr *Addr) (*Conn, error) {

	switch p.mapNetwork(network) {
	case networkMemb, networkMemu:
		// If laddr is not specified then create one with the current
		// epoch in nanoseconds. This value need not be unique.
		if laddr == nil {
			laddr = &Addr{
				Name:    fmt.Sprintf("%d", time.Now().UnixNano()),
				network: network,
			}
		} else {
			laddr.network = network
		}
		if raddr == nil {
			raddr = &Addr{Name: addrLocalhost, network: network}
		} else {
			raddr.network = network
		}
	default:
		return nil, &net.OpError{
			Addr:   raddr,
			Source: laddr,
			Net:    network,
			Op:     "dial",
			Err:    errors.New("unknown network"),
		}
	}

	p.listeners.RLock()
	defer p.listeners.RUnlock()

	if l, ok := p.listeners.cache[raddr.Name]; ok {
		// Update the provided raddr with the actual network type used
		// by the listener.
		raddr.network = l.addr.network
		return l.dial(ctx, network, *laddr, *raddr)
	}

	return nil, &net.OpError{
		Addr:   raddr,
		Source: laddr,
		Net:    network,
		Op:     "dial",
		Err:    errors.New("unknown remote address"),
	}
}
1180712
func clusterDialer(remoteCluster reversetunnel.RemoteSite) auth.DialContext {
	return func(in context.Context, network, _ string) (net.Conn, error) {
		return remoteCluster.DialAuthServer()
	}
}
1366676
func (rc *RegionCoverer) newCoverer() *coverer {
	return &coverer{
		minLevel: maxInt(0, minInt(maxLevel, rc.MinLevel)),
		maxLevel: maxInt(0, minInt(maxLevel, rc.MaxLevel)),
		levelMod: maxInt(1, minInt(3, rc.LevelMod)),
		maxCells: rc.MaxCells,
	}
}
1200098
func NewPostgreSQLCluster(drivername string, connStrings []string) (*Cluster, error) {
	cleanUpDBs := func(dbs []*sql.DB) {
		for _, db := range dbs {
			db.Close()
		}
	}

	dedup := make(map[string]struct{})
	dbs := make([]*sql.DB, 0, len(connStrings))

	if len(connStrings) == 0 {
		return nil, ErrZeroDataSource
	}

	for _, connStr := range connStrings {
		if _, ok := dedup[connStr]; ok {
			cleanUpDBs(dbs)
			return nil, ErrDublicatedDataSource
		}
		dedup[connStr] = struct{}{}

		db, err := sql.Open(drivername, connStr)
		if err != nil {
			cleanUpDBs(dbs)
			return nil, err
		}

		dbs = append(dbs, db)
	}

	cluster := &Cluster{
		dbs: dbs,

		stopCh: make(chan struct{}),
	}

	// electMaster relies on the fact that the value is Stored,
	// so pick the random one
	cluster.setMaster(0, dbs[0])

	cluster.electMaster()

	go cluster.overwatch()

	return cluster, nil
}
310168
func (p *glusterBlockProvisioner) createSecretRef(nameSpace string, secretName string, user string, password string) (*v1.SecretReference, error) {
	var err error
	secret := &v1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: nameSpace,
			Name:      secretName,
		},
		Data: map[string][]byte{
			"node.session.auth.username": []byte(user),
			"node.session.auth.password": []byte(password),
		},
		Type: chapType,
	}

	secretRef := &v1.SecretReference{}
	if secret != nil {
		_, err = p.client.CoreV1().Secrets(nameSpace).Create(secret)
		if err != nil && errors.IsAlreadyExists(err) {

			klog.V(1).Infof("secret %s already exist in namespace %s", secret, nameSpace)
			err = nil
		}
		if err != nil {
			return nil, fmt.Errorf("failed to create secret %s, %v", secret, err)
		}

		if secretRef != nil {
			secretRef.Name = secretName
			klog.V(1).Infof("secret %v and secretRef %v", secret, secretRef)
		}
	} else {
		return nil, fmt.Errorf("secret is nil")

	}
	return secretRef, nil
}
380932
func (j *Job) AddTaskGroup(grp *TaskGroup) *Job {
	j.TaskGroups = append(j.TaskGroups, grp)
	return j
}
960806
func (manager *configManager) WatchCluster(name string, cb func(config *ClusterConfig) error) error {
	manager.Lock()
	defer manager.Unlock()

	if _, present := manager.cbCluster[name]; present {
		return fmt.Errorf("%s already present", name)
	}
	manager.cbCluster[name] = cb
	return nil
}
1345324
func DefaultInstanceGroupVolumeSize(role kops.InstanceGroupRole) (int32, error) {
	switch role {
	case kops.InstanceGroupRoleMaster:
		return DefaultVolumeSizeMaster, nil
	case kops.InstanceGroupRoleNode:
		return DefaultVolumeSizeNode, nil
	case kops.InstanceGroupRoleBastion:
		return DefaultVolumeSizeBastion, nil
	default:
		return -1, fmt.Errorf("unknown InstanceGroup Role %s", role)
	}
}
607229
func (c *Cloud) EnsureLoadBalancer(clusterName string, apiService *v1.Service, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
	annotations := apiService.Annotations
	glog.V(2).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v, %v, %v, %v)",
		clusterName, apiService.Namespace, apiService.Name, c.region, apiService.Spec.LoadBalancerIP, apiService.Spec.Ports, nodes, annotations)

	if apiService.Spec.SessionAffinity != v1.ServiceAffinityNone {
		// ELB supports sticky sessions, but only when configured for HTTP/HTTPS
		return nil, fmt.Errorf("unsupported load balancer affinity: %v", apiService.Spec.SessionAffinity)
	}

	if len(apiService.Spec.Ports) == 0 {
		return nil, fmt.Errorf("requested load balancer with no ports")
	}

	// Figure out what mappings we want on the load balancer
	listeners := []*elb.Listener{}
	portList := getPortSets(annotations[ServiceAnnotationLoadBalancerSSLPorts])
	for _, port := range apiService.Spec.Ports {
		if port.Protocol != v1.ProtocolTCP {
			return nil, fmt.Errorf("Only TCP LoadBalancer is supported for AWS ELB")
		}
		if port.NodePort == 0 {
			glog.Errorf("Ignoring port without NodePort defined: %v", port)
			continue
		}
		listener, err := buildListener(port, annotations, portList)
		if err != nil {
			return nil, err
		}
		listeners = append(listeners, listener)
	}

	if apiService.Spec.LoadBalancerIP != "" {
		return nil, fmt.Errorf("LoadBalancerIP cannot be specified for AWS ELB")
	}

	instances, err := c.getInstancesByNodeNamesCached(nodeNames(nodes))
	if err != nil {
		return nil, err
	}

	sourceRanges, err := service.GetLoadBalancerSourceRanges(apiService)
	if err != nil {
		return nil, err
	}

	// Determine if this is tagged as an Internal ELB
	internalELB := false
	internalAnnotation := apiService.Annotations[ServiceAnnotationLoadBalancerInternal]
	if internalAnnotation != "" {
		if internalAnnotation != "0.0.0.0/0" {
			return nil, fmt.Errorf("annotation %q=%q detected, but the only value supported currently is 0.0.0.0/0", ServiceAnnotationLoadBalancerInternal, internalAnnotation)
		}
		if !service.IsAllowAll(sourceRanges) {
			// TODO: Unify the two annotations
			return nil, fmt.Errorf("source-range annotation cannot be combined with the internal-elb annotation")
		}
		internalELB = true
	}

	// Determine if we need to set the Proxy protocol policy
	proxyProtocol := false
	proxyProtocolAnnotation := apiService.Annotations[ServiceAnnotationLoadBalancerProxyProtocol]
	if proxyProtocolAnnotation != "" {
		if proxyProtocolAnnotation != "*" {
			return nil, fmt.Errorf("annotation %q=%q detected, but the only value supported currently is '*'", ServiceAnnotationLoadBalancerProxyProtocol, proxyProtocolAnnotation)
		}
		proxyProtocol = true
	}

	// Some load balancer attributes are required, so defaults are set. These can be overridden by annotations.
	loadBalancerAttributes := &elb.LoadBalancerAttributes{
		AccessLog:              &elb.AccessLog{Enabled: aws.Bool(false)},
		ConnectionDraining:     &elb.ConnectionDraining{Enabled: aws.Bool(false)},
		ConnectionSettings:     &elb.ConnectionSettings{IdleTimeout: aws.Int64(60)},
		CrossZoneLoadBalancing: &elb.CrossZoneLoadBalancing{Enabled: aws.Bool(false)},
	}

	// Determine if an access log emit interval has been specified
	accessLogEmitIntervalAnnotation := annotations[ServiceAnnotationLoadBalancerAccessLogEmitInterval]
	if accessLogEmitIntervalAnnotation != "" {
		accessLogEmitInterval, parseErr := strconv.ParseInt(accessLogEmitIntervalAnnotation, 10, 64)
		if parseErr != nil {
			return nil, fmt.Errorf("error parsing service annotation: %s=%s",
				ServiceAnnotationLoadBalancerAccessLogEmitInterval,
				accessLogEmitIntervalAnnotation,
			)
		}
		loadBalancerAttributes.AccessLog.EmitInterval = &accessLogEmitInterval
	}

	// Determine if access log enabled/disabled has been specified
	accessLogEnabledAnnotation := annotations[ServiceAnnotationLoadBalancerAccessLogEnabled]
	if accessLogEnabledAnnotation != "" {
		accessLogEnabled, parseErr := strconv.ParseBool(accessLogEnabledAnnotation)
		if parseErr != nil {
			return nil, fmt.Errorf("error parsing service annotation: %s=%s",
				ServiceAnnotationLoadBalancerAccessLogEnabled,
				accessLogEnabledAnnotation,
			)
		}
		loadBalancerAttributes.AccessLog.Enabled = &accessLogEnabled
	}

	// Determine if access log s3 bucket name has been specified
	accessLogS3BucketNameAnnotation := annotations[ServiceAnnotationLoadBalancerAccessLogS3BucketName]
	if accessLogS3BucketNameAnnotation != "" {
		loadBalancerAttributes.AccessLog.S3BucketName = &accessLogS3BucketNameAnnotation
	}

	// Determine if access log s3 bucket prefix has been specified
	accessLogS3BucketPrefixAnnotation := annotations[ServiceAnnotationLoadBalancerAccessLogS3BucketPrefix]
	if accessLogS3BucketPrefixAnnotation != "" {
		loadBalancerAttributes.AccessLog.S3BucketPrefix = &accessLogS3BucketPrefixAnnotation
	}

	// Determine if connection draining enabled/disabled has been specified
	connectionDrainingEnabledAnnotation := annotations[ServiceAnnotationLoadBalancerConnectionDrainingEnabled]
	if connectionDrainingEnabledAnnotation != "" {
		connectionDrainingEnabled, parseErr := strconv.ParseBool(connectionDrainingEnabledAnnotation)
		if parseErr != nil {
			return nil, fmt.Errorf("error parsing service annotation: %s=%s",
				ServiceAnnotationLoadBalancerConnectionDrainingEnabled,
				connectionDrainingEnabledAnnotation,
			)
		}
		loadBalancerAttributes.ConnectionDraining.Enabled = &connectionDrainingEnabled
	}

	// Determine if connection draining timeout has been specified
	connectionDrainingTimeoutAnnotation := annotations[ServiceAnnotationLoadBalancerConnectionDrainingTimeout]
	if connectionDrainingTimeoutAnnotation != "" {
		connectionDrainingTimeout, parseErr := strconv.ParseInt(connectionDrainingTimeoutAnnotation, 10, 64)
		if parseErr != nil {
			return nil, fmt.Errorf("error parsing service annotation: %s=%s",
				ServiceAnnotationLoadBalancerConnectionDrainingTimeout,
				connectionDrainingTimeoutAnnotation,
			)
		}
		loadBalancerAttributes.ConnectionDraining.Timeout = &connectionDrainingTimeout
	}

	// Determine if connection idle timeout has been specified
	connectionIdleTimeoutAnnotation := annotations[ServiceAnnotationLoadBalancerConnectionIdleTimeout]
	if connectionIdleTimeoutAnnotation != "" {
		connectionIdleTimeout, parseErr := strconv.ParseInt(connectionIdleTimeoutAnnotation, 10, 64)
		if parseErr != nil {
			return nil, fmt.Errorf("error parsing service annotation: %s=%s",
				ServiceAnnotationLoadBalancerConnectionIdleTimeout,
				connectionIdleTimeoutAnnotation,
			)
		}
		loadBalancerAttributes.ConnectionSettings.IdleTimeout = &connectionIdleTimeout
	}

	// Determine if cross zone load balancing enabled/disabled has been specified
	crossZoneLoadBalancingEnabledAnnotation := annotations[ServiceAnnotationLoadBalancerCrossZoneLoadBalancingEnabled]
	if crossZoneLoadBalancingEnabledAnnotation != "" {
		crossZoneLoadBalancingEnabled, parseErr := strconv.ParseBool(crossZoneLoadBalancingEnabledAnnotation)
		if parseErr != nil {
			return nil, fmt.Errorf("error parsing service annotation: %s=%s",
				ServiceAnnotationLoadBalancerCrossZoneLoadBalancingEnabled,
				crossZoneLoadBalancingEnabledAnnotation,
			)
		}
		loadBalancerAttributes.CrossZoneLoadBalancing.Enabled = &crossZoneLoadBalancingEnabled
	}

	// Find the subnets that the ELB will live in
	subnetIDs, err := c.findELBSubnets(internalELB)
	if err != nil {
		glog.Error("Error listing subnets in VPC: ", err)
		return nil, err
	}

	// Bail out early if there are no subnets
	if len(subnetIDs) == 0 {
		return nil, fmt.Errorf("could not find any suitable subnets for creating the ELB")
	}

	loadBalancerName := cloudprovider.GetLoadBalancerName(apiService)
	serviceName := types.NamespacedName{Namespace: apiService.Namespace, Name: apiService.Name}

	// Create a security group for the load balancer
	var securityGroupID string
	{
		sgName := "k8s-elb-" + loadBalancerName
		sgDescription := fmt.Sprintf("Security group for Kubernetes ELB %s (%v)", loadBalancerName, serviceName)
		securityGroupID, err = c.ensureSecurityGroup(sgName, sgDescription)
		if err != nil {
			glog.Error("Error creating load balancer security group: ", err)
			return nil, err
		}

		ec2SourceRanges := []*ec2.IpRange{}
		for _, sourceRange := range sourceRanges.StringSlice() {
			ec2SourceRanges = append(ec2SourceRanges, &ec2.IpRange{CidrIp: aws.String(sourceRange)})
		}

		permissions := NewIPPermissionSet()
		for _, port := range apiService.Spec.Ports {
			portInt64 := int64(port.Port)
			protocol := strings.ToLower(string(port.Protocol))

			permission := &ec2.IpPermission{}
			permission.FromPort = &portInt64
			permission.ToPort = &portInt64
			permission.IpRanges = ec2SourceRanges
			permission.IpProtocol = &protocol

			permissions.Insert(permission)
		}

		// Allow ICMP fragmentation packets, important for MTU discovery
		{
			permission := &ec2.IpPermission{
				IpProtocol: aws.String("icmp"),
				FromPort:   aws.Int64(3),
				ToPort:     aws.Int64(4),
				IpRanges:   []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
			}

			permissions.Insert(permission)
		}

		_, err = c.setSecurityGroupIngress(securityGroupID, permissions)
		if err != nil {
			return nil, err
		}
	}
	securityGroupIDs := []string{securityGroupID}

	// Build the load balancer itself
	loadBalancer, err := c.ensureLoadBalancer(
		serviceName,
		loadBalancerName,
		listeners,
		subnetIDs,
		securityGroupIDs,
		internalELB,
		proxyProtocol,
		loadBalancerAttributes,
	)
	if err != nil {
		return nil, err
	}

	err = c.ensureLoadBalancerHealthCheck(loadBalancer, listeners)
	if err != nil {
		return nil, err
	}

	err = c.updateInstanceSecurityGroupsForLoadBalancer(loadBalancer, instances)
	if err != nil {
		glog.Warningf("Error opening ingress rules for the load balancer to the instances: %v", err)
		return nil, err
	}

	err = c.ensureLoadBalancerInstances(orEmpty(loadBalancer.LoadBalancerName), loadBalancer.Instances, instances)
	if err != nil {
		glog.Warningf("Error registering instances with the load balancer: %v", err)
		return nil, err
	}

	glog.V(1).Infof("Loadbalancer %s (%v) has DNS name %s", loadBalancerName, serviceName, orEmpty(loadBalancer.DNSName))

	// TODO: Wait for creation?

	status := toStatus(loadBalancer)
	return status, nil
}
1061791
func NewMasterCountEndpointReconciler(masterCount int, endpointClient corev1client.EndpointsGetter) EndpointReconciler {
	return &masterCountEndpointReconciler{
		masterCount:    masterCount,
		endpointClient: endpointClient,
	}
}
1182393
func (e *Enforcer) HasRoleForUser(name string, role string) bool {
	roles := e.GetRolesForUser(name)

	hasRole := false
	for _, r := range roles {
		if r == role {
			hasRole = true
			break
		}
	}

	return hasRole
}
565450
func (s *Scheduler) EnqueueClusterOperation(ctx context.Context, req *automationpb.EnqueueClusterOperationRequest) (*automationpb.EnqueueClusterOperationResponse, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.state != stateRunning {
		return nil, fmt.Errorf("scheduler is not running. State: %v", s.state)
	}

	if !s.registeredClusterOperations[req.Name] {
		return nil, fmt.Errorf("no ClusterOperation with name: %v is registered", req.Name)
	}

	err := s.validateTaskSpecification(req.Name, req.Parameters)
	if err != nil {
		return nil, err
	}

	clusterOpID := s.idGenerator.GetNextID()
	taskIDGenerator := IDGenerator{}
	initialTask := NewTaskContainerWithSingleTask(req.Name, req.Parameters)
	clusterOp := NewClusterOperationInstance(clusterOpID, initialTask, &taskIDGenerator)

	s.muOpList.Lock()
	s.activeClusterOperations[clusterOpID] = clusterOp.Clone()
	s.muOpList.Unlock()
	s.toBeScheduledClusterOperations <- clusterOp

	return &automationpb.EnqueueClusterOperationResponse{
		Id: clusterOp.Id,
	}, nil
}
58032
func ExpHandler(r metrics.Registry) http.Handler {
	e := exp{sync.Mutex{}, r}
	return http.HandlerFunc(e.expHandler)
}