src/server/pps/server/master.go
func (a *apiServer) master() {
	masterLock := dlock.NewDLock(a.env.GetEtcdClient(), path.Join(a.etcdPrefix, masterLockPath))
	backoff.RetryNotify(func() error {
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		// Use the PPS token to authenticate requests. Note that all requests
		// performed in this function are performed as a cluster admin, so do not
		// pass any unvalidated user input to any requests
		pachClient := a.env.GetPachClient(ctx)
		ctx, err := masterLock.Lock(ctx)
		if err != nil {
			return err
		}
		defer masterLock.Unlock(ctx)
		kubeClient := a.env.GetKubeClient()

		log.Infof("Launching PPS master process")

		pipelineWatcher, err := a.pipelines.ReadOnly(ctx).Watch(watch.WithPrevKV())
		if err != nil {
			return fmt.Errorf("error creating watch: %+v", err)
		}
		defer pipelineWatcher.Close()

		// watchChan will be nil if the Watch call below errors, this means
		// that we won't receive events from k8s and won't be able to detect
		// errors in pods. We could just return that error and retry but that
		// prevents pachyderm from creating pipelines when there's an issue
		// talking to k8s.
		var watchChan <-chan kube_watch.Event
		kubePipelineWatch, err := kubeClient.CoreV1().Pods(a.namespace).Watch(
			metav1.ListOptions{
				LabelSelector: metav1.FormatLabelSelector(metav1.SetAsLabelSelector(
					map[string]string{
						"component": "worker",
					})),
				Watch: true,
			})
		if err != nil {
			log.Errorf("failed to watch kuburnetes pods: %v", err)
		} else {
			watchChan = kubePipelineWatch.ResultChan()
			defer kubePipelineWatch.Stop()
		}

		for {
			select {
			case event := <-pipelineWatcher.Watch():
				if event.Err != nil {
					return fmt.Errorf("event err: %+v", event.Err)
				}
				switch event.Type {
				case watch.EventPut:
					var pipelineName string
					var pipelinePtr pps.EtcdPipelineInfo
					if err := event.Unmarshal(&pipelineName, &pipelinePtr); err != nil {
						return err
					}
					// Retrieve pipelineInfo (and prev pipeline's pipelineInfo) from the
					// spec repo
					var prevPipelinePtr pps.EtcdPipelineInfo
					var pipelineInfo, prevPipelineInfo *pps.PipelineInfo
					if err := a.sudo(pachClient, func(superUserClient *client.APIClient) error {
						var err error
						pipelineInfo, err = ppsutil.GetPipelineInfo(superUserClient, &pipelinePtr, true)
						if err != nil {
							return err
						}

						if event.PrevKey != nil {
							if err := event.UnmarshalPrev(&pipelineName, &prevPipelinePtr); err != nil {
								return err
							}
							prevPipelineInfo, err = ppsutil.GetPipelineInfo(superUserClient, &prevPipelinePtr, true)
							if err != nil {
								return err
							}
						}
						return nil
					}); err != nil {
						return fmt.Errorf("watch event had no pipelineInfo: %v", err)
					}

					// If the pipeline has been stopped, delete workers
					if pipelineInfo.Stopped && pipelineInfo.State != pps.PipelineState_PIPELINE_PAUSED {
						log.Infof("PPS master: deleting workers for pipeline %s (%s)", pipelineName, pipelinePtr.State.String())
						if err := a.deleteWorkersForPipeline(pipelineName); err != nil {
							return err
						}
						if err := a.setPipelineState(pachClient, pipelineInfo, pps.PipelineState_PIPELINE_PAUSED, ""); err != nil {
							return err
						}
					}

					var hasGitInput bool
					pps.VisitInput(pipelineInfo.Input, func(input *pps.Input) {
						if input.Git != nil {
							hasGitInput = true
						}
					})

					// True if the pipeline has been restarted (regardless of any change
					// to the pipeline spec)
					pipelineRestarted := !pipelineInfo.Stopped &&
						event.PrevKey != nil && prevPipelineInfo.Stopped
					// True if auth has been activated or deactivated
					authActivationChanged := (pipelinePtr.AuthToken == "") !=
						(prevPipelinePtr.AuthToken == "")
					// True if the pipeline has been created or updated
					pipelineUpserted := func() bool {
						var prevSpecCommit string
						if prevPipelinePtr.SpecCommit != nil {
							prevSpecCommit = prevPipelinePtr.SpecCommit.ID
						}
						return pipelinePtr.SpecCommit.ID != prevSpecCommit &&
							!pipelineInfo.Stopped
					}()
					if pipelineRestarted || authActivationChanged || pipelineUpserted {
						if (pipelineUpserted || authActivationChanged) && event.PrevKey != nil {
							if err := a.deleteWorkersForPipeline(prevPipelineInfo.Pipeline.Name); err != nil {
								return err
							}
						}
						if (pipelineUpserted || pipelineRestarted) && hasGitInput {
							if err := a.checkOrDeployGithookService(); err != nil {
								return err
							}
						}
						log.Infof("PPS master: creating/updating workers for pipeline %s", pipelineName)
						if err := a.upsertWorkersForPipeline(pipelineInfo); err != nil {
							if err := a.setPipelineState(pachClient, pipelineInfo, pps.PipelineState_PIPELINE_STARTING, fmt.Sprintf("failed to create workers: %s", err.Error())); err != nil {
								return err
							}
							// We return the error here, this causes us to go
							// into backoff and try again from scratch. This
							// means that we'll try creating this pipeline
							// again and also gives a chance for another node,
							// which might actually be able to talk to k8s, to
							// get a chance at creating the workers.
							return err
						}
					}
					if pipelineInfo.State == pps.PipelineState_PIPELINE_RUNNING {
						if err := a.scaleUpWorkersForPipeline(pipelineInfo); err != nil {
							return err
						}
					}
					if pipelineInfo.State == pps.PipelineState_PIPELINE_STANDBY {
						if err := a.scaleDownWorkersForPipeline(pipelineInfo); err != nil {
							return err
						}
					}
				}
			case event := <-watchChan:
				// if we get an error we restart the watch, k8s watches seem to
				// sometimes get stuck in a loop returning events with Type =
				// "" we treat these as errors since otherwise we get an
				// endless stream of them and can't do anything.
				if event.Type == kube_watch.Error || event.Type == "" {
					if kubePipelineWatch != nil {
						kubePipelineWatch.Stop()
					}
					kubePipelineWatch, err = kubeClient.CoreV1().Pods(a.namespace).Watch(
						metav1.ListOptions{
							LabelSelector: metav1.FormatLabelSelector(metav1.SetAsLabelSelector(
								map[string]string{
									"component": "worker",
								})),
							Watch: true,
						})
					if err != nil {
						log.Errorf("failed to watch kuburnetes pods: %v", err)
						watchChan = nil
					} else {
						watchChan = kubePipelineWatch.ResultChan()
						defer kubePipelineWatch.Stop()
					}
				}
				pod, ok := event.Object.(*v1.Pod)
				if !ok {
					continue
				}
				if pod.Status.Phase == v1.PodFailed {
					log.Errorf("pod failed because: %s", pod.Status.Message)
				}
				for _, status := range pod.Status.ContainerStatuses {
					if status.Name == "user" && status.State.Waiting != nil && failures[status.State.Waiting.Reason] {
						if err := a.setPipelineFailure(ctx, pod.ObjectMeta.Annotations["pipelineName"], status.State.Waiting.Message); err != nil {
							return err
						}
					}
				}
			}
		}
	}, backoff.NewInfiniteBackOff(), func(err error, d time.Duration) error {
		for _, c := range a.monitorCancels {
			c()
		}
		a.monitorCancels = make(map[string]func())
		log.Errorf("master: error running the master process: %v; retrying in %v", err, d)
		return nil
	})
}
func (a *apiServer) master() {
	masterLock := dlock.NewDLock(a.env.GetEtcdClient(), path.Join(a.etcdPrefix, masterLockPath))
	backoff.RetryNotify(func() error {
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		// Use the PPS token to authenticate requests. Note that all requests
		// performed in this function are performed as a cluster admin, so do not
		// pass any unvalidated user input to any requests
		pachClient := a.env.GetPachClient(ctx)
		ctx, err := masterLock.Lock(ctx)
		if err != nil {
			return err
		}
		defer masterLock.Unlock(ctx)
		kubeClient := a.env.GetKubeClient()

		log.Infof("Launching PPS master process")

		pipelineWatcher, err := a.pipelines.ReadOnly(ctx).Watch(watch.WithPrevKV())
		if err != nil {
			return fmt.Errorf("error creating watch: %+v", err)
		}
		defer pipelineWatcher.Close()

		// watchChan will be nil if the Watch call below errors, this means
		// that we won't receive events from k8s and won't be able to detect
		// errors in pods. We could just return that error and retry but that
		// prevents pachyderm from creating pipelines when there's an issue
		// talking to k8s.
		var watchChan <-chan kube_watch.Event
		kubePipelineWatch, err := kubeClient.CoreV1().Pods(a.namespace).Watch(
			metav1.ListOptions{
				LabelSelector: metav1.FormatLabelSelector(metav1.SetAsLabelSelector(
					map[string]string{
						"component": "worker",
					})),
				Watch: true,
			})
		if err != nil {
			log.Errorf("failed to watch kuburnetes pods: %v", err)
		} else {
			watchChan = kubePipelineWatch.ResultChan()
			defer kubePipelineWatch.Stop()
		}

		for {
			select {
			case event := <-pipelineWatcher.Watch():
				if event.Err != nil {
					return fmt.Errorf("event err: %+v", event.Err)
				}
				switch event.Type {
				case watch.EventPut:
					var pipelineName string
					var pipelinePtr pps.EtcdPipelineInfo
					if err := event.Unmarshal(&pipelineName, &pipelinePtr); err != nil {
						return err
					}
					// Retrieve pipelineInfo (and prev pipeline's pipelineInfo) from the
					// spec repo
					var prevPipelinePtr pps.EtcdPipelineInfo
					var pipelineInfo, prevPipelineInfo *pps.PipelineInfo
					if err := a.sudo(pachClient, func(superUserClient *client.APIClient) error {
						var err error
						pipelineInfo, err = ppsutil.GetPipelineInfo(superUserClient, &pipelinePtr, true)
						if err != nil {
							return err
						}

						if event.PrevKey != nil {
							if err := event.UnmarshalPrev(&pipelineName, &prevPipelinePtr); err != nil {
								return err
							}
							prevPipelineInfo, err = ppsutil.GetPipelineInfo(superUserClient, &prevPipelinePtr, true)
							if err != nil {
								return err
							}
						}
						return nil
					}); err != nil {
						return fmt.Errorf("watch event had no pipelineInfo: %v", err)
					}

					// If the pipeline has been stopped, delete workers
					if pipelineInfo.Stopped && pipelineInfo.State != pps.PipelineState_PIPELINE_PAUSED {
						log.Infof("PPS master: deleting workers for pipeline %s (%s)", pipelineName, pipelinePtr.State.String())
						if err := a.deleteWorkersForPipeline(pipelineName); err != nil {
							return err
						}
						if err := a.setPipelineState(pachClient, pipelineInfo, pps.PipelineState_PIPELINE_PAUSED, ""); err != nil {
							return err
						}
					}

					var hasGitInput bool
					pps.VisitInput(pipelineInfo.Input, func(input *pps.Input) {
						if input.Git != nil {
							hasGitInput = true
						}
					})

					// True if the pipeline has been restarted (regardless of any change
					// to the pipeline spec)
					pipelineRestarted := !pipelineInfo.Stopped &&
						event.PrevKey != nil && prevPipelineInfo.Stopped
					// True if auth has been activated or deactivated
					authActivationChanged := (pipelinePtr.AuthToken == "") !=
						(prevPipelinePtr.AuthToken == "")
					// True if the pipeline has been created or updated
					pipelineUpserted := func() bool {
						var prevSpecCommit string
						if prevPipelinePtr.SpecCommit != nil {
							prevSpecCommit = prevPipelinePtr.SpecCommit.ID
						}
						return pipelinePtr.SpecCommit.ID != prevSpecCommit &&
							!pipelineInfo.Stopped
					}()
					if pipelineRestarted || authActivationChanged || pipelineUpserted {
						if (pipelineUpserted || authActivationChanged) && event.PrevKey != nil {
							if err := a.deleteWorkersForPipeline(prevPipelineInfo.Pipeline.Name); err != nil {
								return err
							}
						}
						if (pipelineUpserted || pipelineRestarted) && hasGitInput {
							if err := a.checkOrDeployGithookService(); err != nil {
								return err
							}
						}
						log.Infof("PPS master: creating/updating workers for pipeline %s", pipelineName)
						if err := a.upsertWorkersForPipeline(pipelineInfo); err != nil {
							if err := a.setPipelineState(pachClient, pipelineInfo, pps.PipelineState_PIPELINE_STARTING, fmt.Sprintf("failed to create workers: %s", err.Error())); err != nil {
								return err
							}
							// We return the error here, this causes us to go
							// into backoff and try again from scratch. This
							// means that we'll try creating this pipeline
							// again and also gives a chance for another node,
							// which might actually be able to talk to k8s, to
							// get a chance at creating the workers.
							return err
						}
					}
					if pipelineInfo.State == pps.PipelineState_PIPELINE_RUNNING {
						if err := a.scaleUpWorkersForPipeline(pipelineInfo); err != nil {
							return err
						}
					}
					if pipelineInfo.State == pps.PipelineState_PIPELINE_STANDBY {
						if err := a.scaleDownWorkersForPipeline(pipelineInfo); err != nil {
							return err
						}
					}
				}
			case event := <-watchChan:
				// if we get an error we restart the watch, k8s watches seem to
				// sometimes get stuck in a loop returning events with Type =
				// "" we treat these as errors since otherwise we get an
				// endless stream of them and can't do anything.
				if event.Type == kube_watch.Error || event.Type == "" {
					if kubePipelineWatch != nil {
						kubePipelineWatch.Stop()
					}
					kubePipelineWatch, err = kubeClient.CoreV1().Pods(a.namespace).Watch(
						metav1.ListOptions{
							LabelSelector: metav1.FormatLabelSelector(metav1.SetAsLabelSelector(
								map[string]string{
									"component": "worker",
								})),
							Watch: true,
						})
					if err != nil {
						log.Errorf("failed to watch kuburnetes pods: %v", err)
						watchChan = nil
					} else {
						watchChan = kubePipelineWatch.ResultChan()
						defer kubePipelineWatch.Stop()
					}
				}
				pod, ok := event.Object.(*v1.Pod)
				if !ok {
					continue
				}
				if pod.Status.Phase == v1.PodFailed {
					log.Errorf("pod failed because: %s", pod.Status.Message)
				}
				for _, status := range pod.Status.ContainerStatuses {
					if status.Name == "user" && status.State.Waiting != nil && failures[status.State.Waiting.Reason] {
						if err := a.setPipelineFailure(ctx, pod.ObjectMeta.Annotations["pipelineName"], status.State.Waiting.Message); err != nil {
							return err
						}
					}
				}
			}
		}
	}, backoff.NewInfiniteBackOff(), func(err error, d time.Duration) error {
		for _, c := range a.monitorCancels {
			c()
		}
		a.monitorCancels = make(map[string]func())
		log.Errorf("master: error running the master process: %v; retrying in %v", err, d)
		return nil
	})
}
func ( a * apiServer ) master ( ) { masterLock := dlock . NewDLock ( a . env . GetEtcdClient ( ) , path . Join ( a . etcdPrefix , masterLockPath ) ) backoff . RetryNotify ( func ( ) error { ctx , cancel := context . WithCancel ( context . Background ( ) ) defer cancel ( ) // Use the PPS token to authenticate requests. Note that all requests // performed in this function are performed as a cluster admin, so do not // pass any unvalidated user input to any requests pachClient := a . env . GetPachClient ( ctx ) ctx , err := masterLock . Lock ( ctx ) if err != nil { return err } defer masterLock . Unlock ( ctx ) kubeClient := a . env . GetKubeClient ( ) log . Infof ( " " ) pipelineWatcher , err := a . pipelines . ReadOnly ( ctx ) . Watch ( watch . WithPrevKV ( ) ) if err != nil { return fmt . Errorf ( " " , err ) } defer pipelineWatcher . Close ( ) // watchChan will be nil if the Watch call below errors, this means // that we won't receive events from k8s and won't be able to detect // errors in pods. We could just return that error and retry but that // prevents pachyderm from creating pipelines when there's an issue // talking to k8s. var watchChan <- chan kube_watch . Event kubePipelineWatch , err := kubeClient . CoreV1 ( ) . Pods ( a . namespace ) . Watch ( metav1 . ListOptions { LabelSelector : metav1 . FormatLabelSelector ( metav1 . SetAsLabelSelector ( map [ string ] string { " " : " " , } ) ) , Watch : true , } ) if err != nil { log . Errorf ( " " , err ) } else { watchChan = kubePipelineWatch . ResultChan ( ) defer kubePipelineWatch . Stop ( ) } for { select { case event := <- pipelineWatcher . Watch ( ) : if event . Err != nil { return fmt . Errorf ( " " , event . Err ) } switch event . Type { case watch . EventPut : var pipelineName string var pipelinePtr pps . EtcdPipelineInfo if err := event . Unmarshal ( & pipelineName , & pipelinePtr ) ; err != nil { return err } // Retrieve pipelineInfo (and prev pipeline's pipelineInfo) from the // spec repo var prevPipelinePtr pps . EtcdPipelineInfo var pipelineInfo , prevPipelineInfo * pps . PipelineInfo if err := a . sudo ( pachClient , func ( superUserClient * client . APIClient ) error { var err error pipelineInfo , err = ppsutil . GetPipelineInfo ( superUserClient , & pipelinePtr , true ) if err != nil { return err } if event . PrevKey != nil { if err := event . UnmarshalPrev ( & pipelineName , & prevPipelinePtr ) ; err != nil { return err } prevPipelineInfo , err = ppsutil . GetPipelineInfo ( superUserClient , & prevPipelinePtr , true ) if err != nil { return err } } return nil } ) ; err != nil { return fmt . Errorf ( " " , err ) } // If the pipeline has been stopped, delete workers if pipelineInfo . Stopped && pipelineInfo . State != pps . PipelineState_PIPELINE_PAUSED { log . Infof ( " " , pipelineName , pipelinePtr . State . String ( ) ) if err := a . deleteWorkersForPipeline ( pipelineName ) ; err != nil { return err } if err := a . setPipelineState ( pachClient , pipelineInfo , pps . PipelineState_PIPELINE_PAUSED , " " ) ; err != nil { return err } } var hasGitInput bool pps . VisitInput ( pipelineInfo . Input , func ( input * pps . Input ) { if input . Git != nil { hasGitInput = true } } ) // True if the pipeline has been restarted (regardless of any change // to the pipeline spec) pipelineRestarted := ! pipelineInfo . Stopped && event . PrevKey != nil && prevPipelineInfo . Stopped // True if auth has been activated or deactivated authActivationChanged := ( pipelinePtr . AuthToken == " " ) != ( prevPipelinePtr . AuthToken == " " ) // True if the pipeline has been created or updated pipelineUpserted := func ( ) bool { var prevSpecCommit string if prevPipelinePtr . SpecCommit != nil { prevSpecCommit = prevPipelinePtr . SpecCommit . ID } return pipelinePtr . SpecCommit . ID != prevSpecCommit && ! pipelineInfo . Stopped } ( ) if pipelineRestarted || authActivationChanged || pipelineUpserted { if ( pipelineUpserted || authActivationChanged ) && event . PrevKey != nil { if err := a . deleteWorkersForPipeline ( prevPipelineInfo . Pipeline . Name ) ; err != nil { return err } } if ( pipelineUpserted || pipelineRestarted ) && hasGitInput { if err := a . checkOrDeployGithookService ( ) ; err != nil { return err } } log . Infof ( " " , pipelineName ) if err := a . upsertWorkersForPipeline ( pipelineInfo ) ; err != nil { if err := a . setPipelineState ( pachClient , pipelineInfo , pps . PipelineState_PIPELINE_STARTING , fmt . Sprintf ( " " , err . Error ( ) ) ) ; err != nil { return err } // We return the error here, this causes us to go // into backoff and try again from scratch. This // means that we'll try creating this pipeline // again and also gives a chance for another node, // which might actually be able to talk to k8s, to // get a chance at creating the workers. return err } } if pipelineInfo . State == pps . PipelineState_PIPELINE_RUNNING { if err := a . scaleUpWorkersForPipeline ( pipelineInfo ) ; err != nil { return err } } if pipelineInfo . State == pps . PipelineState_PIPELINE_STANDBY { if err := a . scaleDownWorkersForPipeline ( pipelineInfo ) ; err != nil { return err } } } case event := <- watchChan : // if we get an error we restart the watch, k8s watches seem to // sometimes get stuck in a loop returning events with Type = // "" we treat these as errors since otherwise we get an // endless stream of them and can't do anything. if event . Type == kube_watch . Error || event . Type == " " { if kubePipelineWatch != nil { kubePipelineWatch . Stop ( ) } kubePipelineWatch , err = kubeClient . CoreV1 ( ) . Pods ( a . namespace ) . Watch ( metav1 . ListOptions { LabelSelector : metav1 . FormatLabelSelector ( metav1 . SetAsLabelSelector ( map [ string ] string { " " : " " , } ) ) , Watch : true , } ) if err != nil { log . Errorf ( " " , err ) watchChan = nil } else { watchChan = kubePipelineWatch . ResultChan ( ) defer kubePipelineWatch . Stop ( ) } } pod , ok := event . Object . ( * v1 . Pod ) if ! ok { continue } if pod . Status . Phase == v1 . PodFailed { log . Errorf ( " " , pod . Status . Message ) } for _ , status := range pod . Status . ContainerStatuses { if status . Name == " " && status . State . Waiting != nil && failures [ status . State . Waiting . Reason ] { if err := a . setPipelineFailure ( ctx , pod . ObjectMeta . Annotations [ " " ] , status . State . Waiting . Message ) ; err != nil { return err } } } } } } , backoff . NewInfiniteBackOff ( ) , func ( err error , d time . Duration ) error { for _ , c := range a . monitorCancels { c ( ) } a . monitorCancels = make ( map [ string ] func ( ) ) log . Errorf ( " " , err , d ) return nil } ) }
// The master process is responsible for creating/deleting workers as
// pipelines are created/removed.
The master process is responsible for creating / deleting workers as pipelines are created / removed .