[WIP] Implement df daemon

README.md

@@ -22,6 +22,6 @@
 
 ## 架构大图
 ### 产品架构
-![](docs/product-arch.png)
+![](docs/prod-arch.png)
 ### 技术架构
-![](docs/technique-arch.png)
+![](docs/tech-arch.png)

apis/types/dragonfly_version.go

@@ -0,0 +1,70 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package types
+
+import (
+	"github.com/go-openapi/strfmt"
+	"github.com/go-openapi/swag"
+)
+
+// DragonflyVersion Version and build information of Dragonfly components.
+//
+// swagger:model DragonflyVersion
+type DragonflyVersion struct {
+
+	// Dragonfly components's architecture target
+	Arch string `json:"Arch,omitempty"`
+
+	// Build Date of Dragonfly components
+	BuildDate string `json:"BuildDate,omitempty"`
+
+	// Golang runtime version
+	GoVersion string `json:"GoVersion,omitempty"`
+
+	// Dragonfly components's operating system
+	OS string `json:"OS,omitempty"`
+
+	// Git commit when building Dragonfly components
+	Revision string `json:"Revision,omitempty"`
+
+	// Version of Dragonfly components
+	Version string `json:"Version,omitempty"`
+}
+
+// Validate validates this dragonfly version
+func (m *DragonflyVersion) Validate(formats strfmt.Registry) error {
+	return nil
+}
+
+// MarshalBinary interface implementation
+func (m *DragonflyVersion) MarshalBinary() ([]byte, error) {
+	if m == nil {
+		return nil, nil
+	}
+	return swag.WriteJSON(m)
+}
+
+// UnmarshalBinary interface implementation
+func (m *DragonflyVersion) UnmarshalBinary(b []byte) error {
+	var res DragonflyVersion
+	if err := swag.ReadJSON(b, &res); err != nil {
+		return err
+	}
+	*m = res
+	return nil
+}
+

build/protoc.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+SRC="$(cd "$(dirname "$0")/.." && pwd)"
+
+echo "work dir: $SRC"
+
+if protoc -I="$SRC" \
+  --go_out "$SRC" --go_opt paths=source_relative \
+  --go-grpc_out "$SRC" --go-grpc_opt paths=source_relative \
+  "$SRC"/pkg/grpc/base/*.proto \
+  "$SRC"/pkg/grpc/cdnsystem/*.proto \
+  "$SRC"/pkg/grpc/dfdaemon/*.proto \
+  "$SRC"/pkg/grpc/scheduler/*.proto; then
+  echo "generate grpc code successfully"
+else
+  echo "generate grpc code fail"
+fi
+cd "$SRC" && go mod tidy

client/daemon/download_manager.go

@@ -0,0 +1,121 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package daemon
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"time"
+
+	"google.golang.org/grpc"
+
+	dfdaemongrpc "github.com/dragonflyoss/Dragonfly2/pkg/grpc/dfdaemon"
+	"github.com/dragonflyoss/Dragonfly2/pkg/grpc/scheduler"
+)
+
+var _ dfdaemongrpc.DownloaderServer = &downloadManager{}
+
+type DownloadManager interface {
+	ServeGRPC(lis net.Listener) error
+	ServeProxy(lis net.Listener) error
+	Stop() error
+}
+
+type downloadManager struct {
+	peerHost        *scheduler.PeerHost
+	peerTaskManager PeerTaskManager
+
+	grpcServer *grpc.Server
+}
+
+func NewDownloadManager(peerHost *scheduler.PeerHost, peerTaskManager PeerTaskManager) (DownloadManager, error) {
+	mgr := &downloadManager{
+		peerHost:        peerHost,
+		peerTaskManager: peerTaskManager,
+	}
+	return mgr, nil
+}
+
+func (d *downloadManager) ServeGRPC(lis net.Listener) error {
+	s := grpc.NewServer()
+	dfdaemongrpc.RegisterDownloaderServer(s, d)
+	d.grpcServer = s
+	return s.Serve(lis)
+}
+
+func (d *downloadManager) ServeProxy(lis net.Listener) error {
+	// TODO
+	return nil
+}
+
+func (d *downloadManager) Stop() error {
+	d.grpcServer.GracefulStop()
+	// TODO stop proxy
+	return nil
+}
+
+func (d *downloadManager) Download(req *dfdaemongrpc.DownloadRequest, server dfdaemongrpc.Downloader_DownloadServer) error {
+	// init peer task request, peer download request uses different peer id
+	peerTask := &FilePeerTaskRequest{
+		PeerTaskRequest: scheduler.PeerTaskRequest{
+			Url:      req.Url,
+			Filter:   req.Filter,
+			BizId:    req.BizId,
+			UrlMata:  req.UrlMeta,
+			Pid:      d.GenPeerID(),
+			PeerHost: d.peerHost,
+		},
+		Output: req.Output,
+	}
+
+	peerTaskProgress, err := d.peerTaskManager.StartFilePeerTask(context.Background(), peerTask)
+	if err != nil {
+		return err
+	}
+	ctx := server.Context()
+loop:
+	for {
+		select {
+		case p, ok := <-peerTaskProgress:
+			// FIXME is peer task done ?
+			if !ok {
+				break loop
+			}
+			err = server.Send(
+				&dfdaemongrpc.DownloadResult{
+					State:           p.State,
+					TaskId:          p.TaskId,
+					CompletedLength: p.CompletedLength,
+					Done:            p.Done,
+				})
+			if err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	return nil
+}
+
+func (d *downloadManager) GenPeerID() string {
+	// FIXME review peer id format
+	return fmt.Sprintf("%s-%d-%d-%d",
+		d.peerHost.Ip, d.peerHost.Port, os.Getpid(), time.Now().UnixNano())
+}

client/daemon/gc_manager.go

@@ -0,0 +1,70 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package daemon
+
+import (
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+type GC interface {
+	TryGC() (bool, error)
+}
+
+type GCManager interface {
+	Start()
+	Stop()
+}
+
+type gcManager struct {
+	interval time.Duration
+}
+
+var allGCTasks = map[string]GC{}
+
+func RegisterGC(name string, gc GC, interval time.Duration) {
+	allGCTasks[name] = gc
+}
+
+func NewGCManager(interval time.Duration) GCManager {
+	return &gcManager{
+		interval: interval,
+	}
+}
+
+func (g gcManager) Start() {
+	go func() {
+		tick := time.Tick(g.interval)
+		for {
+			select {
+			case <-tick:
+				for name, gc := range allGCTasks {
+					logrus.Infof("start gc %s", name)
+					_, err := gc.TryGC()
+					if err != nil {
+						logrus.Errorf("gc %s error: %s", name, err)
+					}
+				}
+			}
+		}
+	}()
+}
+
+func (g gcManager) Stop() {
+	panic("implement me")
+}