kubeagent

func (c *K8sAgent) otelConfigPath() string {
	daemonsetURI := "/etc/kmagent/agent-daemonset.yaml"
	deploymentURI := "/etc/kmagent/agent-deployment.yaml"
	if c.Cfg.DeploymentMode == "DAEMONSET" {
		return daemonsetURI
	} else {
		return deploymentURI
	}
}

func NewK8sConfig() *K8sConfig {
	config := &K8sConfig{
		ConfigCheckInterval: os.Getenv("KM_CONFIG_CHECK_INTERVAL"),
		APIKey:              os.Getenv("KM_API_KEY"),
		CollectorEndpoint:   os.Getenv("KM_COLLECTOR_ENDPOINT"),
		ConfigMapName:       os.Getenv("CONFIGMAP_NAME"),
		DeploymentMode:      os.Getenv("DEPLOYMENT_MODE"),
		PodNamespace:        os.Getenv("POD_NAMESPACE"),
	}

	if strings.ToUpper(config.DeploymentMode) == "DAEMONSET" {
		config.DeploymentMode = "DAEMONSET"
	} else {
		config.DeploymentMode = "DEPLOYMENT"

	}
	return config
}

func (c *K8sConfig) Validate() error {
	if c.APIKey == "" {
		return fmt.Errorf("KM_API_KEY is required")
	}
	if c.CollectorEndpoint == "" {
		return fmt.Errorf("KM_COLLECTOR_ENDPOINT is required")
	}
	if c.ConfigMapName == "" {
		return fmt.Errorf("CONFIGMAP_NAME is required")
	}
	if c.PodNamespace == "" {
		return fmt.Errorf("POD_NAMESPACE is required")
	}
	return nil
}

kubecfg, err := rest.InClusterConfig()
if err != nil {
	return nil, fmt.Errorf("failed to load in-cluster config: %w", err)
}
logger.Info("loaded in-cluster kubernetes config")

k8sClient, err := kubernetes.NewForConfig(kubecfg)
if err != nil {
	return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
}

func NewK8sAgent(info *AgentInfo) (*K8sAgent, error) {
	zapCfg := zap.NewProductionConfig()
	zapCfg.Level = zap.NewAtomicLevelAt(kmlogger.ParseLogLevel())
	zapLogger, err := zapCfg.Build()
	if err != nil {
		return nil, fmt.Errorf("failed to initialize logger: %w", err)
	}
	logger := zapLogger.Sugar()

	cfg := NewK8sConfig()

	kubecfg, err := rest.InClusterConfig()
	if err != nil {
		return nil, fmt.Errorf("failed to load in-cluster config: %w", err)
	}
	logger.Info("loaded in-cluster kubernetes config")

	k8sClient, err := kubernetes.NewForConfig(kubecfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
	}

	agent := &K8sAgent{
		Cfg:       cfg,
		Logger:    logger,
		K8sClient: k8sClient,
		AgentInfo: *info,
		stopCh:    make(chan struct{}),
	}
	agent.AgentInfo.setEnvForAgentVersion()
	agent.AgentInfo.CollectorVersion = version.GetCollectorVersion()

	logger.Infow("kube agent initialized",
		"version", info.Version,
		"commit", info.CommitSHA,
		"deploymentMode", cfg.DeploymentMode,
	)
	return agent, nil
}

// handleSignals sets up a signal handler to gracefully shut down the agent.
func handleSignals(cancelFunc context.CancelFunc, agent *k8sagent.K8sAgent) {
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		sig := <-sigChan
		agent.Logger.Warnf("Received signal %s, initiating shutdown...", sig)
		cancelFunc()
		agent.Stop()
	}()
}

func main() {
	// Set up the main context for the application
	appCtx, cancelAppCtx := context.WithCancel(context.Background())
	defer cancelAppCtx()

	agent, err := k8sagent.NewK8sAgent(&k8sagent.AgentInfo{Version: version, CommitSHA: commit})
	if err != nil {
		log.Fatal(err)
	}

	// Handle OS signals for graceful shutdown.
	handleSignals(cancelAppCtx, agent)

	if err = agent.StartAgent(appCtx); err != nil {
		agent.Logger.Errorf("agent could not be started with current config : %s", err.Error())
	}

	agent.AwaitShutdown()

	defer func() {
		// Ensure logger is synced before exit to flush buffered logs.
		if syncErr := agent.Logger.Sync(); syncErr != nil && syncErr.Error() != "sync /dev/stdout: invalid argument" {
			agent.Logger.Warnf("Failed to sync logger: %v\n", syncErr)
		}
	}()
}

kubectl apply -f deployment.yaml

zapCfg := zap.NewProductionConfig()
zapCfg.Level = zap.NewAtomicLevelAt(kmlogger.ParseLogLevel())

{
  "level": "info",
  "ts": "2024-03-15T10:30:45.123Z",
  "msg": "kube agent initialized",
  "version": "0.1.0",
  "commit": "abc123",
  "deploymentMode": "DEPLOYMENT"
}

var (
	version = "0.1.0"
	commit  = "none"
)

// setEnvForAgentVersion sets agent version on env for otel processor
func (r *AgentInfo) setEnvForAgentVersion() {
	os.Setenv("KM_AGENT_VERSION", r.Version)
}

// Stop stops the underlying collector.
func (a *K8sAgent) Stop() {
	a.Logger.Info("stopping collector agent")
	close(a.stopCh)
	a.wg.Wait()
	a.stopInternalCollector()
	a.Logger.Info("collector agent stopped")
}

kubectl get pods -n kloudmate -l app=kubeagent
kubectl describe pod -n kloudmate -l app=kubeagent

kubectl logs -n kloudmate -l app=kubeagent --tail=100 -f

# Check ConfigMap
kubectl get configmap -n kloudmate kmagent-config -o yaml

# Check Secret
kubectl get secret -n kloudmate kloudmate-secret -o yaml

# Verify environment variables
kubectl exec -n kloudmate deployment/kubeagent -- env | grep KM_

kubectl edit configmap -n kloudmate kmagent-config

livenessProbe:
  httpGet:
    path: /health
    port: 13133
  initialDelaySeconds: 30
  periodSeconds: 10
readinessProbe:
  httpGet:
    path: /ready
    port: 13133
  initialDelaySeconds: 5
  periodSeconds: 5