// InitSlave sets replication master and position, and waits for the
// reparent_journal table entry up to context timeout
173195
func (agent *ActionAgent) InitSlave(ctx context.Context, parent *topodatapb.TabletAlias, position string, timeCreatedNS int64) error {
	if err := agent.lock(ctx); err != nil {
		return err
	}
	defer agent.unlock()

	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return err
	}
	ti, err := agent.TopoServer.GetTablet(ctx, parent)
	if err != nil {
		return err
	}

	agent.setSlaveStopped(false)

	// If using semi-sync, we need to enable it before connecting to master.
	// If we were a master type, we need to switch back to replica settings.
	// Otherwise we won't be able to commit anything.
	tt := agent.Tablet().Type
	if tt == topodatapb.TabletType_MASTER {
		tt = topodatapb.TabletType_REPLICA
	}
	if err := agent.fixSemiSync(tt); err != nil {
		return err
	}

	if err := agent.MysqlDaemon.SetSlavePosition(ctx, pos); err != nil {
		return err
	}
	if err := agent.MysqlDaemon.SetMaster(ctx, topoproto.MysqlHostname(ti.Tablet), int(topoproto.MysqlPort(ti.Tablet)), false /* slaveStopBefore */, true /* slaveStartAfter */); err != nil {
		return err
	}
	agent.initReplication = true

	// If we were a master type, switch our type to replica.  This
	// is used on the old master when using InitShardMaster with
	// -force, and the new master is different from the old master.
	if agent.Tablet().Type == topodatapb.TabletType_MASTER {
		if _, err := topotools.ChangeType(ctx, agent.TopoServer, agent.TabletAlias, topodatapb.TabletType_REPLICA); err != nil {
			return err
		}

		if err := agent.refreshTablet(ctx, "InitSlave"); err != nil {
			return err
		}
	}

	// wait until we get the replicated row, or our context times out
	return agent.MysqlDaemon.WaitForReparentJournal(ctx, timeCreatedNS)
}
1148336
func SocketInit(so *wsevent.Client) error {
	loggedIn := so.Token != nil

	if loggedIn {
		hooks.AfterConnect(socket.AuthServer, so)

		steamid := so.Token.Claims.(*chelpers.TF2StadiumClaims).SteamID

		player, err := player.GetPlayerBySteamID(steamid)
		if err != nil {
			return fmt.Errorf("Couldn't find player record for %s", steamid)
		}

		hooks.AfterConnectLoggedIn(so, player)
	} else {
		hooks.AfterConnect(socket.UnauthServer, so)
		so.EmitJSON(helpers.NewRequest("playerSettings", "{}"))
		so.EmitJSON(helpers.NewRequest("playerProfile", "{}"))
	}

	so.EmitJSON(helpers.NewRequest("socketInitialized", "{}"))

	return nil
}
635655
func (s *MockSchemaSyncer) Init(ctx context.Context) error {
	s.globalVerCh = make(chan clientv3.WatchResponse, 1)
	s.mockSession = make(chan struct{}, 1)
	return nil
}
1334393
func InitRepository(repoPath string, bare bool) error {
	os.MkdirAll(repoPath, os.ModePerm)

	cmd := NewCommand("init")
	if bare {
		cmd.AddArguments("--bare")
	}
	_, err := cmd.RunInDir(repoPath)
	return err
}
641078
func (c *V1) EnsureInit() error {
	c.initOnce.Do(c.init)
	return c.aclCheckerInitErr
}
659866
func OnInit(ctx context.Context, manifest *plugin.Manifest) error {
	fmt.Println("conn_ip_example init called")
	fmt.Println("read cfg in init", manifest.SysVars["conn_ip_example_test_variable"].Value)
	return nil
}
1351523
func RecordStartSlaveAction(cleaner *Cleaner, tablet *topodatapb.Tablet) {
	cleaner.Record(StartSlaveActionName, topoproto.TabletAliasString(tablet.Alias), func(ctx context.Context, wr *Wrangler) error {
		return wr.TabletManagerClient().StartSlave(ctx, tablet)
	})
}
102936
func ProtoToSlaveStatus(s *replicationdatapb.Status) SlaveStatus {
	pos, err := DecodePosition(s.Position)
	if err != nil {
		panic(vterrors.Wrapf(err, "cannot decode Position"))
	}
	return SlaveStatus{
		Position:            pos,
		SlaveIORunning:      s.SlaveIoRunning,
		SlaveSQLRunning:     s.SlaveSqlRunning,
		SecondsBehindMaster: uint(s.SecondsBehindMaster),
		MasterHost:          s.MasterHost,
		MasterPort:          int(s.MasterPort),
		MasterConnectRetry:  int(s.MasterConnectRetry),
	}
}
499279
func (rp *Ringpop) init() error {
	if rp.channel == nil {
		return errors.New("Missing channel")
	}

	address, err := rp.address()
	if err != nil {
		return err
	}

	// early initialization of statter before registering listeners that might
	// fire and try to stat
	rp.stats.hostport = genStatsHostport(address)
	rp.stats.prefix = fmt.Sprintf("ringpop.%s", rp.stats.hostport)
	rp.stats.keys = make(map[string]string)

	rp.subChannel = rp.channel.GetSubChannel("ringpop", tchannel.Isolated)
	rp.registerHandlers()

	rp.node = swim.NewNode(rp.config.App, address, rp.subChannel, &swim.Options{
		StateTimeouts:     rp.config.StateTimeouts,
		Clock:             rp.clock,
		LabelLimits:       rp.config.LabelLimits,
		InitialLabels:     rp.config.InitialLabels,
		SelfEvict:         rp.config.SelfEvict,
		RequiresAppInPing: rp.config.RequiresAppInPing,
	})
	rp.node.AddListener(rp)

	rp.ring = hashring.New(farm.Fingerprint32, rp.configHashRing.ReplicaPoints)
	rp.ring.AddListener(rp)

	// add all members present in the membership of the node on startup.
	for _, member := range rp.node.GetReachableMembers() {
		rp.ring.AddMembers(member)
	}

	rp.forwarder = forward.NewForwarder(rp, rp.subChannel)
	rp.forwarder.AddListener(rp)

	rp.startTimers()
	rp.setState(initialized)

	return nil
}
824417
func (tr *Repo) InitSnapshot() error {
	if tr.Root == nil {
		return ErrNotLoaded{Role: data.CanonicalRootRole}
	}
	root, err := tr.Root.ToSigned()
	if err != nil {
		return err
	}

	if _, ok := tr.Targets[data.CanonicalTargetsRole]; !ok {
		return ErrNotLoaded{Role: data.CanonicalTargetsRole}
	}
	targets, err := tr.Targets[data.CanonicalTargetsRole].ToSigned()
	if err != nil {
		return err
	}
	snapshot, err := data.NewSnapshot(root, targets)
	if err != nil {
		return err
	}
	tr.Snapshot = snapshot
	return nil
}
532055
func (vm *Machine) Init() {
	vm.State = NewState()
	vm.Positions = append(vm.Positions, Position{State: NewState()})
	vm.Imperial = false
	vm.AbsoluteMove = true
	vm.AbsoluteArc = false
	vm.MovePlane = PlaneXY
	vm.MaxArcDeviation = 0.002
	vm.MinArcLineLength = 0.01
	vm.IgnoreBlockDelete = false
}
657437
func (s *BaseGenerator) Init() {
	s.Position = vm.Position{State: vm.NewState()}
}
1303506
func (agent *ActionAgent) PromoteSlave(ctx context.Context) (string, error) {
	if err := agent.lock(ctx); err != nil {
		return "", err
	}
	defer agent.unlock()

	pos, err := agent.MysqlDaemon.PromoteSlave(agent.hookExtraEnv())
	if err != nil {
		return "", err
	}

	// If using semi-sync, we need to enable it before going read-write.
	if err := agent.fixSemiSync(topodatapb.TabletType_MASTER); err != nil {
		return "", err
	}

	// Set the server read-write
	startTime := time.Now()
	if err := agent.MysqlDaemon.SetReadOnly(false); err != nil {
		return "", err
	}
	agent.setExternallyReparentedTime(startTime)

	if _, err := topotools.ChangeType(ctx, agent.TopoServer, agent.TabletAlias, topodatapb.TabletType_MASTER); err != nil {
		return "", err
	}

	if err := agent.refreshTablet(ctx, "PromoteSlave"); err != nil {
		return "", err
	}

	return mysql.EncodePosition(pos), nil
}
786696
func (c *Repair) runOnce() error {
	// TODO: (per smarterclayton) if Get() or ListServices() is a weak consistency read,
	// or if they are executed against different leaders,
	// the ordering guarantee required to ensure no IP is allocated twice is violated.
	// ListServices must return a ResourceVersion higher than the etcd index Get triggers,
	// and the release code must not release services that have had IPs allocated but not yet been created
	// See #8295

	// If etcd server is not running we should wait for some time and fail only then. This is particularly
	// important when we start apiserver and etcd at the same time.
	var snapshot *api.RangeAllocation
	err := wait.PollImmediate(time.Second, 10*time.Second, func() (bool, error) {
		var err error
		snapshot, err = c.alloc.Get()
		return err == nil, err
	})
	if err != nil {
		return fmt.Errorf("unable to refresh the service IP block: %v", err)
	}
	// If not yet initialized.
	if snapshot.Range == "" {
		snapshot.Range = c.network.String()
	}
	// Create an allocator because it is easy to use.
	stored, err := ipallocator.NewFromSnapshot(snapshot)
	if err != nil {
		return fmt.Errorf("unable to rebuild allocator from snapshot: %v", err)
	}

	// We explicitly send no resource version, since the resource version
	// of 'snapshot' is from a different collection, it's not comparable to
	// the service collection. The caching layer keeps per-collection RVs,
	// and this is proper, since in theory the collections could be hosted
	// in separate etcd (or even non-etcd) instances.
	list, err := c.serviceClient.Services(metav1.NamespaceAll).List(metav1.ListOptions{})
	if err != nil {
		return fmt.Errorf("unable to refresh the service IP block: %v", err)
	}

	rebuilt := ipallocator.NewCIDRRange(c.network)
	// Check every Service's ClusterIP, and rebuild the state as we think it should be.
	for _, svc := range list.Items {
		if !helper.IsServiceIPSet(&svc) {
			// didn't need a cluster IP
			continue
		}
		ip := net.ParseIP(svc.Spec.ClusterIP)
		if ip == nil {
			// cluster IP is corrupt
			c.recorder.Eventf(&svc, v1.EventTypeWarning, "ClusterIPNotValid", "Cluster IP %s is not a valid IP; please recreate service", svc.Spec.ClusterIP)
			runtime.HandleError(fmt.Errorf("the cluster IP %s for service %s/%s is not a valid IP; please recreate", svc.Spec.ClusterIP, svc.Name, svc.Namespace))
			continue
		}
		// mark it as in-use
		switch err := rebuilt.Allocate(ip); err {
		case nil:
			if stored.Has(ip) {
				// remove it from the old set, so we can find leaks
				stored.Release(ip)
			} else {
				// cluster IP doesn't seem to be allocated
				c.recorder.Eventf(&svc, v1.EventTypeWarning, "ClusterIPNotAllocated", "Cluster IP %s is not allocated; repairing", ip)
				runtime.HandleError(fmt.Errorf("the cluster IP %s for service %s/%s is not allocated; repairing", ip, svc.Name, svc.Namespace))
			}
			delete(c.leaks, ip.String()) // it is used, so it can't be leaked
		case ipallocator.ErrAllocated:
			// cluster IP is duplicate
			c.recorder.Eventf(&svc, v1.EventTypeWarning, "ClusterIPAlreadyAllocated", "Cluster IP %s was assigned to multiple services; please recreate service", ip)
			runtime.HandleError(fmt.Errorf("the cluster IP %s for service %s/%s was assigned to multiple services; please recreate", ip, svc.Name, svc.Namespace))
		case err.(*ipallocator.ErrNotInRange):
			// cluster IP is out of range
			c.recorder.Eventf(&svc, v1.EventTypeWarning, "ClusterIPOutOfRange", "Cluster IP %s is not within the service CIDR %s; please recreate service", ip, c.network)
			runtime.HandleError(fmt.Errorf("the cluster IP %s for service %s/%s is not within the service CIDR %s; please recreate", ip, svc.Name, svc.Namespace, c.network))
		case ipallocator.ErrFull:
			// somehow we are out of IPs
			c.recorder.Eventf(&svc, v1.EventTypeWarning, "ServiceCIDRFull", "Service CIDR %s is full; you must widen the CIDR in order to create new services", c.network)
			return fmt.Errorf("the service CIDR %s is full; you must widen the CIDR in order to create new services", c.network)
		default:
			c.recorder.Eventf(&svc, v1.EventTypeWarning, "UnknownError", "Unable to allocate cluster IP %s due to an unknown error", ip)
			return fmt.Errorf("unable to allocate cluster IP %s for service %s/%s due to an unknown error, exiting: %v", ip, svc.Name, svc.Namespace, err)
		}
	}

	// Check for IPs that are left in the old set.  They appear to have been leaked.
	stored.ForEach(func(ip net.IP) {
		count, found := c.leaks[ip.String()]
		switch {
		case !found:
			// flag it to be cleaned up after any races (hopefully) are gone
			runtime.HandleError(fmt.Errorf("the cluster IP %s may have leaked: flagging for later clean up", ip))
			count = numRepairsBeforeLeakCleanup - 1
			fallthrough
		case count > 0:
			// pretend it is still in use until count expires
			c.leaks[ip.String()] = count - 1
			if err := rebuilt.Allocate(ip); err != nil {
				runtime.HandleError(fmt.Errorf("the cluster IP %s may have leaked, but can not be allocated: %v", ip, err))
			}
		default:
			// do not add it to the rebuilt set, which means it will be available for reuse
			runtime.HandleError(fmt.Errorf("the cluster IP %s appears to have leaked: cleaning up", ip))
		}
	})

	// Blast the rebuilt state into storage.
	if err := rebuilt.Snapshot(snapshot); err != nil {
		return fmt.Errorf("unable to snapshot the updated service IP allocations: %v", err)
	}
	if err := c.alloc.CreateOrUpdate(snapshot); err != nil {
		if errors.IsConflict(err) {
			return err
		}
		return fmt.Errorf("unable to persist the updated service IP allocations: %v", err)
	}
	return nil
}
454238
func (process *TeleportProcess) initAuthService() error {
	var err error

	cfg := process.Config

	// Initialize the storage back-ends for keys, events and records
	b, err := process.initAuthStorage()
	if err != nil {
		return trace.Wrap(err)
	}
	process.backend = b

	// create the audit log, which will be consuming (and recording) all events
	// and recording all sessions.
	if cfg.Auth.NoAudit {
		// this is for teleconsole
		process.auditLog = events.NewDiscardAuditLog()

		warningMessage := "Warning: Teleport audit and session recording have been " +
			"turned off. This is dangerous, you will not be able to view audit events " +
			"or save and playback recorded sessions."
		process.Warn(warningMessage)
	} else {
		// check if session recording has been disabled. note, we will continue
		// logging audit events, we just won't record sessions.
		recordSessions := true
		if cfg.Auth.ClusterConfig.GetSessionRecording() == services.RecordOff {
			recordSessions = false

			warningMessage := "Warning: Teleport session recording have been turned off. " +
				"This is dangerous, you will not be able to save and playback sessions."
			process.Warn(warningMessage)
		}

		auditConfig := cfg.Auth.ClusterConfig.GetAuditConfig()
		uploadHandler, err := initUploadHandler(auditConfig)
		if err != nil {
			if !trace.IsNotFound(err) {
				return trace.Wrap(err)
			}
		}

		externalLog, err := initExternalLog(auditConfig)
		if err != nil {
			if !trace.IsNotFound(err) {
				return trace.Wrap(err)
			}
		}

		auditServiceConfig := events.AuditLogConfig{
			Context:        process.ExitContext(),
			DataDir:        filepath.Join(cfg.DataDir, teleport.LogsDir),
			RecordSessions: recordSessions,
			ServerID:       cfg.HostUUID,
			UploadHandler:  uploadHandler,
			ExternalLog:    externalLog,
		}
		auditServiceConfig.UID, auditServiceConfig.GID, err = adminCreds()
		if err != nil {
			return trace.Wrap(err)
		}
		process.auditLog, err = events.NewAuditLog(auditServiceConfig)
		if err != nil {
			return trace.Wrap(err)
		}
	}

	// first, create the AuthServer
	authServer, err := auth.Init(auth.InitConfig{
		Backend:              b,
		Authority:            cfg.Keygen,
		ClusterConfiguration: cfg.ClusterConfiguration,
		ClusterConfig:        cfg.Auth.ClusterConfig,
		ClusterName:          cfg.Auth.ClusterName,
		AuthServiceName:      cfg.Hostname,
		DataDir:              cfg.DataDir,
		HostUUID:             cfg.HostUUID,
		NodeName:             cfg.Hostname,
		Authorities:          cfg.Auth.Authorities,
		ReverseTunnels:       cfg.ReverseTunnels,
		Trust:                cfg.Trust,
		Presence:             cfg.Presence,
		Events:               cfg.Events,
		Provisioner:          cfg.Provisioner,
		Identity:             cfg.Identity,
		Access:               cfg.Access,
		StaticTokens:         cfg.Auth.StaticTokens,
		Roles:                cfg.Auth.Roles,
		AuthPreference:       cfg.Auth.Preference,
		OIDCConnectors:       cfg.OIDCConnectors,
		AuditLog:             process.auditLog,
		CipherSuites:         cfg.CipherSuites,
	})
	if err != nil {
		return trace.Wrap(err)
	}

	process.setLocalAuth(authServer)

	connector, err := process.connectToAuthService(teleport.RoleAdmin)
	if err != nil {
		return trace.Wrap(err)
	}

	// second, create the API Server: it's actually a collection of API servers,
	// each serving requests for a "role" which is assigned to every connected
	// client based on their certificate (user, server, admin, etc)
	sessionService, err := session.New(b)
	if err != nil {
		return trace.Wrap(err)
	}
	authorizer, err := auth.NewAuthorizer(authServer.Access, authServer.Identity, authServer.Trust)
	if err != nil {
		return trace.Wrap(err)
	}
	apiConf := &auth.APIConfig{
		AuthServer:     authServer,
		SessionService: sessionService,
		Authorizer:     authorizer,
		AuditLog:       process.auditLog,
	}

	var authCache auth.AuthCache
	if process.Config.CachePolicy.Enabled {
		cache, err := process.newAccessCache(accessCacheConfig{
			services:  authServer.AuthServices,
			setup:     cache.ForAuth,
			cacheName: []string{teleport.ComponentAuth},
			inMemory:  true,
			events:    true,
		})
		if err != nil {
			return trace.Wrap(err)
		}
		authCache = cache
	} else {
		authCache = authServer.AuthServices
	}
	authServer.SetCache(authCache)

	log := logrus.WithFields(logrus.Fields{
		trace.Component: teleport.Component(teleport.ComponentAuth, process.id),
	})

	// Register TLS endpoint of the auth service
	tlsConfig, err := connector.ServerIdentity.TLSConfig(cfg.CipherSuites)
	if err != nil {
		return trace.Wrap(err)
	}
	tlsServer, err := auth.NewTLSServer(auth.TLSServerConfig{
		TLS:           tlsConfig,
		APIConfig:     *apiConf,
		LimiterConfig: cfg.Auth.Limiter,
		AccessPoint:   authCache,
		Component:     teleport.Component(teleport.ComponentAuth, process.id),
	})
	if err != nil {
		return trace.Wrap(err)
	}
	// auth server listens on SSH and TLS, reusing the same socket
	listener, err := process.importOrCreateListener(teleport.ComponentAuth, cfg.Auth.SSHAddr.Addr)
	if err != nil {
		log.Errorf("PID: %v Failed to bind to address %v: %v, exiting.", os.Getpid(), cfg.Auth.SSHAddr.Addr, err)
		return trace.Wrap(err)
	}
	// clean up unused descriptors passed for proxy, but not used by it
	warnOnErr(process.closeImportedDescriptors(teleport.ComponentAuth))
	if cfg.Auth.EnableProxyProtocol {
		log.Infof("Starting Auth service with PROXY protocol support.")
	}
	mux, err := multiplexer.New(multiplexer.Config{
		EnableProxyProtocol: cfg.Auth.EnableProxyProtocol,
		Listener:            listener,
		ID:                  teleport.Component(process.id),
	})
	if err != nil {
		listener.Close()
		return trace.Wrap(err)
	}
	go mux.Serve()
	process.RegisterCriticalFunc("auth.tls", func() error {
		utils.Consolef(cfg.Console, teleport.ComponentAuth, "Auth service is starting on %v.", cfg.Auth.SSHAddr.Addr)

		// since tlsServer.Serve is a blocking call, we emit this even right before
		// the service has started
		process.BroadcastEvent(Event{Name: AuthTLSReady, Payload: nil})
		err := tlsServer.Serve(mux.TLS())
		if err != nil && err != http.ErrServerClosed {
			log.Warningf("TLS server exited with error: %v.", err)
		}
		return nil
	})
	process.RegisterFunc("auth.heartbeat.broadcast", func() error {
		// Heart beat auth server presence, this is not the best place for this
		// logic, consolidate it into auth package later
		connector, err := process.connectToAuthService(teleport.RoleAdmin)
		if err != nil {
			return trace.Wrap(err)
		}
		// External integrations rely on this event:
		process.BroadcastEvent(Event{Name: AuthIdentityEvent, Payload: connector})
		process.onExit("auth.broadcast", func(payload interface{}) {
			connector.Close()
		})
		return nil
	})

	// figure out server public address
	authAddr := cfg.Auth.SSHAddr.Addr
	host, port, err := net.SplitHostPort(authAddr)
	if err != nil {
		return trace.Wrap(err)
	}
	// advertise-ip is explicitly set:
	if process.Config.AdvertiseIP != "" {
		ahost, aport, err := utils.ParseAdvertiseAddr(process.Config.AdvertiseIP)
		if err != nil {
			return trace.Wrap(err)
		}
		// if port is not set in the advertise addr, use the default one
		if aport == "" {
			aport = port
		}
		authAddr = fmt.Sprintf("%v:%v", ahost, aport)
	} else {
		// advertise-ip is not set, while the CA is listening on 0.0.0.0? lets try
		// to guess the 'advertise ip' then:
		if net.ParseIP(host).IsUnspecified() {
			ip, err := utils.GuessHostIP()
			if err != nil {
				log.Warn(err)
			} else {
				authAddr = net.JoinHostPort(ip.String(), port)
			}
		}
		log.Warnf("Configuration setting auth_service/advertise_ip is not set. guessing %v.", authAddr)
	}

	heartbeat, err := srv.NewHeartbeat(srv.HeartbeatConfig{
		Mode:      srv.HeartbeatModeAuth,
		Context:   process.ExitContext(),
		Component: teleport.ComponentAuth,
		Announcer: authServer,
		GetServerInfo: func() (services.Server, error) {
			srv := services.ServerV2{
				Kind:    services.KindAuthServer,
				Version: services.V2,
				Metadata: services.Metadata{
					Namespace: defaults.Namespace,
					Name:      process.Config.HostUUID,
				},
				Spec: services.ServerSpecV2{
					Addr:     authAddr,
					Hostname: process.Config.Hostname,
				},
			}
			state, err := process.storage.GetState(teleport.RoleAdmin)
			if err != nil {
				if !trace.IsNotFound(err) {
					log.Warningf("Failed to get rotation state: %v.", err)
					return nil, trace.Wrap(err)
				}
			} else {
				srv.Spec.Rotation = state.Spec.Rotation
			}
			srv.SetTTL(process, defaults.ServerAnnounceTTL)
			return &srv, nil
		},
		KeepAlivePeriod: defaults.ServerKeepAliveTTL,
		AnnouncePeriod:  defaults.ServerAnnounceTTL/2 + utils.RandomDuration(defaults.ServerAnnounceTTL/10),
		CheckPeriod:     defaults.HeartbeatCheckPeriod,
		ServerTTL:       defaults.ServerAnnounceTTL,
	})
	if err != nil {
		return trace.Wrap(err)
	}
	process.RegisterFunc("auth.heartbeat", heartbeat.Run)
	// execute this when process is asked to exit:
	process.onExit("auth.shutdown", func(payload interface{}) {
		// The listeners have to be closed here, because if shutdown
		// was called before the start of the http server,
		// the http server would have not started tracking the listeners
		// and http.Shutdown will do nothing.
		if mux != nil {
			warnOnErr(mux.Close())
		}
		if listener != nil {
			warnOnErr(listener.Close())
		}
		if payload == nil {
			log.Info("Shutting down immediately.")
			warnOnErr(tlsServer.Close())
		} else {
			log.Info("Shutting down gracefully.")
			ctx := payloadContext(payload)
			warnOnErr(tlsServer.Shutdown(ctx))
		}
		log.Info("Exited.")
	})
	return nil
}
1121848
func (w *Writer) Init(target querypb.Target) error {
	if !w.enabled {
		return nil
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	log.Info("Initializing heartbeat table.")
	w.dbName = sqlescape.EscapeID(w.dbconfigs.SidecarDBName.Get())
	w.keyspaceShard = fmt.Sprintf("%s:%s", target.Keyspace, target.Shard)
	err := w.initializeTables(w.dbconfigs.DbaWithDB())
	if err != nil {
		w.recordError(err)
		return err
	}

	return nil
}
45202
func (agent *ActionAgent) StopSlave(ctx context.Context) error {
	if err := agent.lock(ctx); err != nil {
		return err
	}
	defer agent.unlock()

	return agent.stopSlaveLocked(ctx)
}