func()

in x-pack/auditbeat/module/system/socket/socket_linux.go [282:465]
133 lines of code
32 McCabe index (conditional complexity)

func (m *MetricSet) Setup() (err error) {
	m.log.Infof("Setting up %s for kernel %s", fullName, kernelVersion)

	//
	// Validate that tracefs / debugfs is present and kprobes are available
	//
	var traceFS *tracing.TraceFS
	if m.config.TraceFSPath == nil {
		if err := tracing.IsTraceFSAvailable(); err != nil {
			m.log.Debugf("tracefs/debugfs not found. Attempting to mount")
			for _, mount := range defaultMounts {
				if err = mount.mount(); err != nil {
					m.log.Debugf("Mount %s returned %v", mount, err)
					continue
				}
				if tracing.IsTraceFSAvailable() != nil {
					m.log.Warnf("Mounted %s but no kprobes available", mount, err)
					mount.unmount()
					continue
				}
				m.log.Debugf("Mounted %s", mount)
				m.mountedFS = mount
				break
			}
		}
		traceFS, err = tracing.NewTraceFS()
	} else {
		traceFS, err = tracing.NewTraceFSWithPath(*m.config.TraceFSPath)
	}
	if err != nil {
		return fmt.Errorf("tracefs/debugfs is not mounted or not writeable: %w", err)
	}

	//
	// Setup initial template variables
	//
	m.templateVars.Update(baseTemplateVars)
	m.templateVars.Update(archVariables)

	//
	// Detect IPv6 support
	//

	hasIPv6, err := detectIPv6()
	if err != nil {
		m.log.Debugf("Error detecting IPv6 support: %v", err)
		hasIPv6 = false
	}
	m.log.Debugf("IPv6 supported: %v", hasIPv6)
	if m.config.EnableIPv6 != nil {
		if *m.config.EnableIPv6 && !hasIPv6 {
			return errors.New("requested IPv6 support but IPv6 is disabled in the system")
		}
		hasIPv6 = *m.config.EnableIPv6
	}
	m.log.Debugf("IPv6 enabled: %v", hasIPv6)
	m.templateVars["HAS_IPV6"] = hasIPv6

	//
	// Create probe installer
	//
	extra := WithNoOp()
	if m.config.DevelopmentMode {
		extra = WithFilterPort(22)
	}
	m.installer = newProbeInstaller(traceFS,
		WithGroup(groupName),
		WithTemplates(m.templateVars),
		extra)
	defer func() {
		if err != nil {
			m.installer.UninstallInstalled()
		}
	}()

	//
	// remove dangling KProbes from terminated Auditbeat processes.
	// Not a fatal error if they can't be removed.
	//
	if err = m.installer.UninstallIf(isDeadAuditbeat); err != nil {
		m.log.Debugf("Removing existing probes from terminated instances: %+v", err)
	}

	//
	// remove existing Auditbeat KProbes that match the current PID.
	//
	if err = m.installer.UninstallIf(isThisAuditbeat); err != nil {
		return fmt.Errorf("unable to delete existing KProbes for group %s: %w", groupName, err)
	}

	//
	// Load available kernel functions for tracing
	//
	functions, err := LoadTracingFunctions(traceFS)
	if err != nil {
		m.log.Debugf("Can't load available_tracing_functions. Using alternative. err=%v", err)
	}

	//
	// Resolve function names from alternatives
	//
	for varName, alternatives := range functionAlternatives {
		if exists, _ := m.templateVars.HasKey(varName); exists {
			return fmt.Errorf("variable %s overwrites existing key", varName)
		}
		found := false
		var selected string
		for _, selected = range alternatives {
			if found = m.isKernelFunctionAvailable(selected, functions); found {
				break
			}
		}
		if !found {
			return fmt.Errorf("none of the required functions for %s is found. One of %v is required", varName, alternatives)
		}
		if m.isDebug {
			m.log.Debugf("Selected kernel function %s for %s", selected, varName)
		}
		m.templateVars[varName] = selected
	}

	//
	// Make sure all the required kernel functions are available
	//
	for _, probeDef := range getKProbes(hasIPv6) {
		probeDef = probeDef.ApplyTemplate(m.templateVars)
		name := probeDef.Probe.Address
		if !m.isKernelFunctionAvailable(name, functions) {
			return fmt.Errorf("required function '%s' is not available for tracing in the current kernel (%s)", name, kernelVersion)
		}
	}

	//
	// Guess all the required parameters
	//
	if err = guess.GuessAll(m.installer,
		guess.Context{
			Log:     m.log,
			Vars:    m.templateVars,
			Timeout: m.config.GuessTimeout,
		}); err != nil {
		return fmt.Errorf("unable to guess one or more required parameters: %w", err)
	}

	if m.isDebug {
		names := make([]string, 0, len(m.templateVars))
		for name := range m.templateVars {
			names = append(names, name)
		}
		sort.Strings(names)
		m.log.Debugf("%d template variables in use:", len(m.templateVars))
		for _, key := range names {
			m.log.Debugf("  %s = %v", key, m.templateVars[key])
		}
	}

	//
	// Create perf channel
	//
	m.perfChannel, err = tracing.NewPerfChannel(
		tracing.WithBufferSize(m.config.PerfQueueSize),
		tracing.WithErrBufferSize(m.config.ErrQueueSize),
		tracing.WithLostBufferSize(m.config.LostQueueSize),
		tracing.WithRingSizeExponent(m.config.RingSizeExp),
		tracing.WithTID(perf.AllThreads),
		tracing.WithTimestamp())
	if err != nil {
		return fmt.Errorf("unable to create perf channel: %w", err)
	}

	//
	// Register Kprobes
	//
	for _, probeDef := range getKProbes(hasIPv6) {
		format, decoder, err := m.installer.Install(probeDef)
		if err != nil {
			return fmt.Errorf("unable to register probe %s: %w", probeDef.Probe.String(), err)
		}
		if err = m.perfChannel.MonitorProbe(format, decoder); err != nil {
			return fmt.Errorf("unable to monitor probe %s: %w", probeDef.Probe.String(), err)
		}
	}
	return nil
}