[Feature request] 导出查询日志

pkg/nftset_utils/handler.go

@@ -121,7 +121,7 @@ func (h *NftSetHandler) AddElems(es ...netip.Prefix) error {
 		if set.Interval {
 			start := e.Masked().Addr()
 			elems = append(elems, nftables.SetElement{Key: start.AsSlice(), IntervalEnd: false})
-			
+
 			end := netipx.PrefixLastIP(e).Next() // may be invalid if end is overflowed
 			if end.IsValid() {
 				elems = append(elems, nftables.SetElement{Key: end.AsSlice(), IntervalEnd: true})

pkg/upstream/transport/conn_quic.go

@@ -120,4 +120,4 @@ func (ote *quicReservedExchanger) WithdrawReserved() {
 	s := ote.stream
 	s.CancelRead(_DOQ_REQUEST_CANCELLED)
 	s.CancelWrite(_DOQ_REQUEST_CANCELLED)
-}
+}

plugin/enabled_plugins.go

@@ -71,4 +71,7 @@ import (
 	_ "github.com/IrineSistiana/mosdns/v5/plugin/server/quic_server"
 	_ "github.com/IrineSistiana/mosdns/v5/plugin/server/tcp_server"
 	_ "github.com/IrineSistiana/mosdns/v5/plugin/server/udp_server"
+
+	// statistics
+	_ "github.com/IrineSistiana/mosdns/v5/plugin/statistics/simple"
 )

plugin/executable/arbitrary/arbitrary.go

@@ -23,12 +23,14 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"os"
+	"strings"
+
 	"github.com/IrineSistiana/mosdns/v5/coremain"
 	"github.com/IrineSistiana/mosdns/v5/pkg/query_context"
 	"github.com/IrineSistiana/mosdns/v5/pkg/zone_file"
 	"github.com/IrineSistiana/mosdns/v5/plugin/executable/sequence"
-	"os"
-	"strings"
+	"github.com/IrineSistiana/mosdns/v5/plugin/statistics"
 )
 
 const PluginType = "arbitrary"
@@ -72,6 +74,7 @@ func NewArbitrary(args *Args) (*Arbitrary, error) {
 func (a *Arbitrary) Exec(_ context.Context, qCtx *query_context.Context) error {
 	if r := a.m.Reply(qCtx.Q()); r != nil {
 		qCtx.SetResponse(r)
+		qCtx.StoreValue(statistics.ArbitraryStoreKey, true)
 	}
 	return nil
 }

plugin/executable/cache/cache.go

@@ -38,6 +38,7 @@ import (
 	"github.com/IrineSistiana/mosdns/v5/pkg/query_context"
 	"github.com/IrineSistiana/mosdns/v5/pkg/utils"
 	"github.com/IrineSistiana/mosdns/v5/plugin/executable/sequence"
+	"github.com/IrineSistiana/mosdns/v5/plugin/statistics"
 	"github.com/go-chi/chi/v5"
 	"github.com/klauspost/compress/gzip"
 	"github.com/miekg/dns"
@@ -202,6 +203,7 @@ func (c *Cache) Exec(ctx context.Context, qCtx *query_context.Context, next sequ
 	}
 	if cachedResp != nil { // cache hit
 		c.hitTotal.Inc()
+		qCtx.StoreValue(statistics.CacaheStoreKey, cachedResp.Id)
 		cachedResp.Id = q.Id // change msg id
 		qCtx.SetResponse(cachedResp)
 	}

plugin/executable/forward/forward.go

@@ -33,6 +33,7 @@ import (
 	"github.com/IrineSistiana/mosdns/v5/pkg/upstream"
 	"github.com/IrineSistiana/mosdns/v5/pkg/utils"
 	"github.com/IrineSistiana/mosdns/v5/plugin/executable/sequence"
+	"github.com/IrineSistiana/mosdns/v5/plugin/statistics"
 	"github.com/miekg/dns"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
@@ -193,11 +194,12 @@ func (f *Forward) RegisterMetricsTo(r prometheus.Registerer) error {
 }
 
 func (f *Forward) Exec(ctx context.Context, qCtx *query_context.Context) (err error) {
-	r, err := f.exchange(ctx, qCtx, f.us)
+	r, name, err := f.exchange(ctx, qCtx, f.us)
 	if err != nil {
 		return err
 	}
 	qCtx.SetResponse(r)
+	qCtx.StoreValue(statistics.ForwardStoreKey, name)
 	return nil
 }
 
@@ -216,11 +218,12 @@ func (f *Forward) QuickConfigureExec(args string) (any, error) {
 		}
 	}
 	var execFunc sequence.ExecutableFunc = func(ctx context.Context, qCtx *query_context.Context) error {
-		r, err := f.exchange(ctx, qCtx, us)
+		r, name, err := f.exchange(ctx, qCtx, us)
 		if err != nil {
 			return err
 		}
 		qCtx.SetResponse(r)
+		qCtx.StoreValue(statistics.ForwardStoreKey, name)
 		return nil
 	}
 	return execFunc, nil
@@ -233,14 +236,14 @@ func (f *Forward) Close() error {
 	return nil
 }
 
-func (f *Forward) exchange(ctx context.Context, qCtx *query_context.Context, us []*upstreamWrapper) (*dns.Msg, error) {
+func (f *Forward) exchange(ctx context.Context, qCtx *query_context.Context, us []*upstreamWrapper) (*dns.Msg, string, error) {
 	if len(us) == 0 {
-		return nil, errors.New("no upstream to exchange")
+		return nil, "", errors.New("no upstream to exchange")
 	}
 
 	queryPayload, err := pool.PackBuffer(qCtx.Q())
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	defer pool.ReleaseBuf(queryPayload)
 
@@ -255,6 +258,7 @@ func (f *Forward) exchange(ctx context.Context, qCtx *query_context.Context, us
 	type res struct {
 		r   *dns.Msg
 		err error
+		u   string
 	}
 
 	resChan := make(chan res)
@@ -291,7 +295,7 @@ func (f *Forward) exchange(ctx context.Context, qCtx *query_context.Context, us
 				}
 			}
 			select {
-			case resChan <- res{r: r, err: err}:
+			case resChan <- res{r: r, u: u.name(), err: err}:
 			case <-done:
 			}
 		}(qCtx.Id(), qCtx.QQuestion())
@@ -309,12 +313,12 @@ func (f *Forward) exchange(ctx context.Context, qCtx *query_context.Context, us
 			if i < concurrent-1 && r.Rcode != dns.RcodeSuccess && r.Rcode != dns.RcodeNameError {
 				continue
 			}
-			return r, nil
+			return r, res.u, nil
 		case <-ctx.Done():
-			return nil, context.Cause(ctx)
+			return nil, "", context.Cause(ctx)
 		}
 	}
-	return nil, errors.New("all upstream servers failed")
+	return nil, "", errors.New("all upstream servers failed")
 }
 
 func quickSetup(bq sequence.BQ, s string) (any, error) {

plugin/executable/hosts/hosts.go

@@ -23,13 +23,15 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"os"
+
 	"github.com/IrineSistiana/mosdns/v5/coremain"
 	"github.com/IrineSistiana/mosdns/v5/pkg/hosts"
 	"github.com/IrineSistiana/mosdns/v5/pkg/matcher/domain"
 	"github.com/IrineSistiana/mosdns/v5/pkg/query_context"
 	"github.com/IrineSistiana/mosdns/v5/plugin/executable/sequence"
+	"github.com/IrineSistiana/mosdns/v5/plugin/statistics"
 	"github.com/miekg/dns"
-	"os"
 )
 
 const PluginType = "hosts"
@@ -84,6 +86,7 @@ func (h *Hosts) Exec(_ context.Context, qCtx *query_context.Context) error {
 	r := h.h.LookupMsg(qCtx.Q())
 	if r != nil {
 		qCtx.SetResponse(r)
+		qCtx.StoreValue(statistics.HostStoreKey, true)
 	}
 	return nil
 }

plugin/executable/sequence/fallback/fallback.go

@@ -29,6 +29,7 @@ import (
 	"github.com/IrineSistiana/mosdns/v5/pkg/pool"
 	"github.com/IrineSistiana/mosdns/v5/pkg/query_context"
 	"github.com/IrineSistiana/mosdns/v5/plugin/executable/sequence"
+	"github.com/IrineSistiana/mosdns/v5/plugin/statistics"
 	"github.com/miekg/dns"
 	"go.uber.org/zap"
 )
@@ -131,6 +132,7 @@ func (f *fallback) doFallback(ctx context.Context, qCtx *query_context.Context)
 			close(primDone)
 			respChan <- r
 		}
+		qCtx.StoreValue(statistics.FallbackStoreKey, uint(1))
 	}()
 
 	// Secondary goroutine.
@@ -168,6 +170,7 @@ func (f *fallback) doFallback(ctx context.Context, qCtx *query_context.Context)
 			}
 		}
 		respChan <- r
+		qCtx.StoreValue(statistics.FallbackStoreKey, uint(2))
 	}()
 
 	for i := 0; i < 2; i++ {

plugin/statistics/key.go

@@ -0,0 +1,11 @@
+package statistics
+
+import "github.com/IrineSistiana/mosdns/v5/pkg/query_context"
+
+var (
+	ArbitraryStoreKey = query_context.RegKey()
+	HostStoreKey      = query_context.RegKey()
+	ForwardStoreKey   = query_context.RegKey()
+	FallbackStoreKey  = query_context.RegKey()
+	CacaheStoreKey    = query_context.RegKey()
+)

plugin/statistics/simple/api.go

@@ -0,0 +1,52 @@
+package simple
+
+import (
+	"container/list"
+	"net/http"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"go.uber.org/zap"
+)
+
+func (c *simpleServer) Api() *chi.Mux {
+	r := chi.NewRouter()
+	r.Get("/statistics", c.statistics)
+	return r
+}
+
+func (c *simpleServer) statistics(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Access-Control-Allow-Origin", "*")
+	w.WriteHeader(200)
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		w.Write([]byte("Unsupported connection method"))
+		return
+	}
+
+	var elem *list.Element // maybe nil
+	var lastElem *list.Element
+	for {
+		select {
+		case <-r.Context().Done():
+			return
+		default:
+			if lastElem != nil {
+				elem = lastElem.Next()
+			} else {
+				elem = c.backend.Front()
+			}
+
+			if elem != nil {
+				lastElem = elem
+				if _, err := w.Write(append(elem.Value.([]byte), '\n')); err != nil {
+					c.logger.Warn("Http write error", zap.Error(err))
+					return
+				}
+				flusher.Flush()
+			} else {
+				time.Sleep(time.Second)
+			}
+		}
+	}
+}

plugin/statistics/simple/exec.go

@@ -0,0 +1,74 @@
+package simple
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/IrineSistiana/mosdns/v5/pkg/query_context"
+	"github.com/IrineSistiana/mosdns/v5/plugin/executable/sequence"
+	"go.uber.org/zap"
+)
+
+var (
+	httpClient = &http.Client{
+		Transport: &http.Transport{
+			TLSHandshakeTimeout: time.Second * 5,
+			IdleConnTimeout:     time.Second * 60,
+			MaxIdleConns:        5,
+			MaxIdleConnsPerHost: 5,
+			DisableCompression:  true,
+		},
+	}
+)
+
+func (m *simpleServer) Exec(ctx context.Context, qCtx *query_context.Context, next sequence.ChainWalker) (err error) {
+	record := recordPool.Get().(*record)
+	defer record.release()
+
+	record.SetQuery(qCtx)
+
+	err = next.ExecNext(ctx, qCtx)
+	record.Err = err
+
+	if r := qCtx.R(); r != nil {
+		record.SetResp(qCtx)
+	}
+
+	bd, _ := json.Marshal(record)
+	go m.push(bd)
+	return
+}
+
+func (m *simpleServer) push(d []byte) {
+	if m.backend.Len() > m.args.Size {
+		_ = m.backend.Remove(m.backend.Front())
+	}
+	m.backend.PushBack(d)
+	m.send(d)
+}
+
+func (m *simpleServer) send(d []byte) (err error) {
+	if m.args.WebHook == "" {
+		return
+	}
+	resp, err := httpClient.Post(m.args.WebHook, "application/json", bytes.NewBuffer(d))
+	if err != nil {
+		m.logger.Debug("HTTP request failed", zap.Error(err))
+		return
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		m.logger.Debug("Failed to read response", zap.Error(err))
+		return
+	}
+	defer resp.Body.Close() // Close the response body.
+	if resp.StatusCode > 299 {
+		m.logger.Debug("WebHook sent incorrect data", zap.String("status", resp.Status), zap.String("body", string(body)))
+	}
+	return
+}