Clair介绍和源码分析

version: '2'
services:
  postgresql:
    image: /libary/postgres:0.1
    restart: always
    ports:
      - 5432:5432
    volumes:
      - /docker/postgresql/data:/var/lib/postgresql/data
  clair:
    image: libary/clair:0.2
    depends_on:
      - postgresql
    ports:
      - 6060:6060
      - 6061:6061
    environment:
      - POSTGRESQL_HOST=postgresql

func main() {
	...
	// 加载配置文件
	config, err := config.Load(*flagConfigPath)
	...
	// Enable CPU Profiling if specified
	if *flagCPUProfilePath != "" {
		defer stopCPUProfiling(startCPUProfiling(*flagCPUProfilePath))
	}
	// 启动clair
	clair.Boot(config)
}

func Boot(config *config.Config) {
	...
	// 连接后端DB，默认配置为pgsql
	db, err := database.Open(config.Database)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	// 启动Notifier服务，clair实现了webhook notifier
	st.Begin()
	go notifier.Run(config.Notifier, db, st)

	// 启动clair的Rest API 服务
	st.Begin()
	go api.Run(config.API, &context.RouteContext{db, config.API}, st)
	
	// 启动clair的健康检查端口
	st.Begin()
	go api.RunHealth(config.API, &context.RouteContext{db, config.API}, st)

	// 启动定期的Updater服务，clair实现了fetcher updater
	st.Begin()
	go updater.Run(config.Updater, db, st)
	...
}

func Run(config *config.APIConfig, ctx *context.RouteContext, st *utils.Stopper) {
	...
	srv := &graceful.Server{
		Timeout:          0,    // Already handled by our TimeOut middleware
		NoSignalHandling: true, // We want to use our own Stopper
		Server: &http.Server{
			Addr:      ":" + strconv.Itoa(config.Port),
			TLSConfig: tlsConfig,
			Handler:   http.TimeoutHandler(newAPIHandler(ctx), config.Timeout, timeoutResponse),
		},
	}

	listenAndServeWithStopper(srv, st, config.CertFile, config.KeyFile)
	...
}

func newAPIHandler(ctx *context.RouteContext) http.Handler {
	router := make(router)
	router["/v1"] = v1.NewRouter(ctx)
	return router
}

// NewRouter creates an HTTP router for version 1 of the Clair API.
func NewRouter(ctx *context.RouteContext) *httprouter.Router {
	router := httprouter.New()

	// Layers
	router.POST("/layers", context.HTTPHandler(postLayer, ctx))
	router.GET("/layers/:layerName", context.HTTPHandler(getLayer, ctx))
	router.DELETE("/layers/:layerName", context.HTTPHandler(deleteLayer, ctx))

	// Namespaces
	router.GET("/namespaces", context.HTTPHandler(getNamespaces, ctx))

	// Vulnerabilities
	router.GET("/namespaces/:namespaceName/vulnerabilities", context.HTTPHandler(getVulnerabilities, ctx))
	router.POST("/namespaces/:namespaceName/vulnerabilities", context.HTTPHandler(postVulnerability, ctx))
	router.GET("/namespaces/:namespaceName/vulnerabilities/:vulnerabilityName", context.HTTPHandler(getVulnerability, ctx))
	router.PUT("/namespaces/:namespaceName/vulnerabilities/:vulnerabilityName", context.HTTPHandler(putVulnerability, ctx))
	router.DELETE("/namespaces/:namespaceName/vulnerabilities/:vulnerabilityName", context.HTTPHandler(deleteVulnerability, ctx))

	// Fixes
	router.GET("/namespaces/:namespaceName/vulnerabilities/:vulnerabilityName/fixes", context.HTTPHandler(getFixes, ctx))
	router.PUT("/namespaces/:namespaceName/vulnerabilities/:vulnerabilityName/fixes/:fixName", context.HTTPHandler(putFix, ctx))
	router.DELETE("/namespaces/:namespaceName/vulnerabilities/:vulnerabilityName/fixes/:fixName", context.HTTPHandler(deleteFix, ctx))

	// Notifications
	router.GET("/notifications/:notificationName", context.HTTPHandler(getNotification, ctx))
	router.DELETE("/notifications/:notificationName", context.HTTPHandler(deleteNotification, ctx))

	// Metrics
	router.GET("/metrics", context.HTTPHandler(getMetrics, ctx))

	return router
}

func postLayer(w http.ResponseWriter, r *http.Request, p httprouter.Params, ctx *context.RouteContext) (string, int) {
	...
	err = worker.Process(ctx.Store, request.Layer.Format, request.Layer.Name, request.Layer.ParentName, request.Layer.Path, request.Layer.Headers)
	...
}

func Process(datastore database.Datastore, imageFormat, name, parentName, path string, headers map[string]string) error {
	...
	// Check to see if the layer is already in the database.
	layer, err := datastore.FindLayer(name, false, false)
	if err != nil && err != cerrors.ErrNotFound {
		return err
	}

	if err == cerrors.ErrNotFound {
		// New layer case.
		layer = database.Layer{Name: name, EngineVersion: Version}

		// Retrieve the parent if it has one.
		// We need to get it with its Features in order to diff them.
		if parentName != "" {
			parent, err := datastore.FindLayer(parentName, true, false)
			if err != nil && err != cerrors.ErrNotFound {
				return err
			}
			if err == cerrors.ErrNotFound {
				log.Warningf("layer %s: the parent layer (%s) is unknown. it must be processed first", name,
					parentName)
				return ErrParentUnknown
			}
			layer.Parent = &parent
		}
	} else {
		// The layer is already in the database, check if we need to update it.
		if layer.EngineVersion >= Version {
			log.Debugf(`layer %s: layer content has already been processed in the past with engine %d.
        Current engine is %d. skipping analysis`, name, layer.EngineVersion, Version)
			return nil
		}

		log.Debugf(`layer %s: layer content has been analyzed in the past with engine %d. Current
      engine is %d. analyzing again`, name, layer.EngineVersion, Version)
	}

	// Analyze the content.
	layer.Namespace, layer.Features, err = detectContent(imageFormat, name, path, headers, layer.Parent)
	if err != nil {
		return err
	}

	return datastore.InsertLayer(layer)
}

// detectContent downloads a layer's archive and extracts its Namespace and Features.
func detectContent(imageFormat, name, path string, headers map[string]string, parent *database.Layer) (namespace *database.Namespace, featureVersions []database.FeatureVersion, err error) {

	// Detect Data
	data, err := detectors.DetectData(imageFormat, path, headers, append(detectors.GetRequiredFilesFeatures(), detectors.GetRequiredFilesNamespace()...), maxFileSize)
	...

	// Detect namespace.
	namespace = detectNamespace(name, data, parent)

	// Detect features.
	featureVersions, err = detectFeatureVersions(name, data, namespace, parent)
	...
}

func DetectData(format, path string, headers map[string]string, toExtract []string, maxFileSize int64) (data map[string][]byte, err error) {
	...
	if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") {
		// Create a new HTTP request object.
		request, err := http.NewRequest("GET", path, nil)
		...
		layerReader = r.Body
	} else {
		layerReader, err = os.Open(path)
		...
	}
	defer layerReader.Close()

	for _, detector := range dataDetectors {
		if detector.Supported(path, format) {
			data, err = detector.Detect(layerReader, toExtract, maxFileSize)
			...
		}
	}

	...
}

func (detector *DockerDataDetector) Detect(layerReader io.ReadCloser, toExtract []string, maxFileSize int64) (map[string][]byte, error) {
	return utils.SelectivelyExtractArchive(layerReader, "", toExtract, maxFileSize)
}

func (detector *ACIDataDetector) Detect(layerReader io.ReadCloser, toExtract []string, maxFileSize int64) (map[string][]byte, error) {
	return utils.SelectivelyExtractArchive(layerReader, "rootfs/", toExtract, maxFileSize)
}

// SelectivelyExtractArchive extracts the specified files and folders
// from targz data read from the given reader and store them in a map indexed by file paths
func SelectivelyExtractArchive(r io.Reader, prefix string, toExtract []string, maxFileSize int64) (map[string][]byte, error) {
	data := make(map[string][]byte)
	...
	return data, nil
}