mirror of
https://github.com/caddyserver/caddy.git
synced 2024-11-24 20:46:51 +08:00
events: Implement event system (#4912)
Co-authored-by: Matt Holt <mholt@users.noreply.github.com>
This commit is contained in:
parent
68d8ac9802
commit
d4d8bbcfc6
68
context.go
68
context.go
|
@ -37,9 +37,10 @@ import (
|
|||
// not actually need to do this).
|
||||
type Context struct {
|
||||
context.Context
|
||||
moduleInstances map[string][]any
|
||||
moduleInstances map[string][]Module
|
||||
cfg *Config
|
||||
cleanupFuncs []func()
|
||||
ancestry []Module
|
||||
}
|
||||
|
||||
// NewContext provides a new context derived from the given
|
||||
|
@ -51,7 +52,7 @@ type Context struct {
|
|||
// modules which are loaded will be properly unloaded.
|
||||
// See standard library context package's documentation.
|
||||
func NewContext(ctx Context) (Context, context.CancelFunc) {
|
||||
newCtx := Context{moduleInstances: make(map[string][]any), cfg: ctx.cfg}
|
||||
newCtx := Context{moduleInstances: make(map[string][]Module), cfg: ctx.cfg}
|
||||
c, cancel := context.WithCancel(ctx.Context)
|
||||
wrappedCancel := func() {
|
||||
cancel()
|
||||
|
@ -90,15 +91,15 @@ func (ctx *Context) OnCancel(f func()) {
|
|||
// ModuleMap may be used in place of map[string]json.RawMessage. The return value's
|
||||
// underlying type mirrors the input field's type:
|
||||
//
|
||||
// json.RawMessage => any
|
||||
// []json.RawMessage => []any
|
||||
// [][]json.RawMessage => [][]any
|
||||
// map[string]json.RawMessage => map[string]any
|
||||
// []map[string]json.RawMessage => []map[string]any
|
||||
// json.RawMessage => any
|
||||
// []json.RawMessage => []any
|
||||
// [][]json.RawMessage => [][]any
|
||||
// map[string]json.RawMessage => map[string]any
|
||||
// []map[string]json.RawMessage => []map[string]any
|
||||
//
|
||||
// The field must have a "caddy" struct tag in this format:
|
||||
//
|
||||
// caddy:"key1=val1 key2=val2"
|
||||
// caddy:"key1=val1 key2=val2"
|
||||
//
|
||||
// To load modules, a "namespace" key is required. For example, to load modules
|
||||
// in the "http.handlers" namespace, you'd put: `namespace=http.handlers` in the
|
||||
|
@ -115,7 +116,7 @@ func (ctx *Context) OnCancel(f func()) {
|
|||
// meaning the key containing the module's name that is defined inline with the module
|
||||
// itself. You must specify the inline key in a struct tag, along with the namespace:
|
||||
//
|
||||
// caddy:"namespace=http.handlers inline_key=handler"
|
||||
// caddy:"namespace=http.handlers inline_key=handler"
|
||||
//
|
||||
// This will look for a key/value pair like `"handler": "..."` in the json.RawMessage
|
||||
// in order to know the module name.
|
||||
|
@ -301,17 +302,17 @@ func (ctx Context) loadModuleMap(namespace string, val reflect.Value) (map[strin
|
|||
// like from embedded scripts, etc.
|
||||
func (ctx Context) LoadModuleByID(id string, rawMsg json.RawMessage) (any, error) {
|
||||
modulesMu.RLock()
|
||||
mod, ok := modules[id]
|
||||
modInfo, ok := modules[id]
|
||||
modulesMu.RUnlock()
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unknown module: %s", id)
|
||||
}
|
||||
|
||||
if mod.New == nil {
|
||||
return nil, fmt.Errorf("module '%s' has no constructor", mod.ID)
|
||||
if modInfo.New == nil {
|
||||
return nil, fmt.Errorf("module '%s' has no constructor", modInfo.ID)
|
||||
}
|
||||
|
||||
val := mod.New().(any)
|
||||
val := modInfo.New()
|
||||
|
||||
// value must be a pointer for unmarshaling into concrete type, even if
|
||||
// the module's concrete type is a slice or map; New() *should* return
|
||||
|
@ -327,7 +328,7 @@ func (ctx Context) LoadModuleByID(id string, rawMsg json.RawMessage) (any, error
|
|||
if len(rawMsg) > 0 {
|
||||
err := strictUnmarshalJSON(rawMsg, &val)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decoding module config: %s: %v", mod, err)
|
||||
return nil, fmt.Errorf("decoding module config: %s: %v", modInfo, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -340,6 +341,8 @@ func (ctx Context) LoadModuleByID(id string, rawMsg json.RawMessage) (any, error
|
|||
return nil, fmt.Errorf("module value cannot be null")
|
||||
}
|
||||
|
||||
ctx.ancestry = append(ctx.ancestry, val)
|
||||
|
||||
if prov, ok := val.(Provisioner); ok {
|
||||
err := prov.Provision(ctx)
|
||||
if err != nil {
|
||||
|
@ -351,7 +354,7 @@ func (ctx Context) LoadModuleByID(id string, rawMsg json.RawMessage) (any, error
|
|||
err = fmt.Errorf("%v; additionally, cleanup: %v", err, err2)
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("provision %s: %v", mod, err)
|
||||
return nil, fmt.Errorf("provision %s: %v", modInfo, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -365,7 +368,7 @@ func (ctx Context) LoadModuleByID(id string, rawMsg json.RawMessage) (any, error
|
|||
err = fmt.Errorf("%v; additionally, cleanup: %v", err, err2)
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("%s: invalid configuration: %v", mod, err)
|
||||
return nil, fmt.Errorf("%s: invalid configuration: %v", modInfo, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -439,8 +442,10 @@ func (ctx Context) Storage() certmagic.Storage {
|
|||
return ctx.cfg.storage
|
||||
}
|
||||
|
||||
// TODO: aw man, can I please change this?
|
||||
// Logger returns a logger that can be used by mod.
|
||||
func (ctx Context) Logger(mod Module) *zap.Logger {
|
||||
// TODO: if mod is nil, use ctx.Module() instead...
|
||||
if ctx.cfg == nil {
|
||||
// often the case in tests; just use a dev logger
|
||||
l, err := zap.NewDevelopment()
|
||||
|
@ -451,3 +456,34 @@ func (ctx Context) Logger(mod Module) *zap.Logger {
|
|||
}
|
||||
return ctx.cfg.Logging.Logger(mod)
|
||||
}
|
||||
|
||||
// TODO: use this
|
||||
// // Logger returns a logger that can be used by the current module.
|
||||
// func (ctx Context) Log() *zap.Logger {
|
||||
// if ctx.cfg == nil {
|
||||
// // often the case in tests; just use a dev logger
|
||||
// l, err := zap.NewDevelopment()
|
||||
// if err != nil {
|
||||
// panic("config missing, unable to create dev logger: " + err.Error())
|
||||
// }
|
||||
// return l
|
||||
// }
|
||||
// return ctx.cfg.Logging.Logger(ctx.Module())
|
||||
// }
|
||||
|
||||
// Modules returns the lineage of modules that this context provisioned,
|
||||
// with the most recent/current module being last in the list.
|
||||
func (ctx Context) Modules() []Module {
|
||||
mods := make([]Module, len(ctx.ancestry))
|
||||
copy(mods, ctx.ancestry)
|
||||
return mods
|
||||
}
|
||||
|
||||
// Module returns the current module, or the most recent one
|
||||
// provisioned by the context.
|
||||
func (ctx Context) Module() Module {
|
||||
if len(ctx.ancestry) == 0 {
|
||||
return nil
|
||||
}
|
||||
return ctx.ancestry[len(ctx.ancestry)-1]
|
||||
}
|
||||
|
|
2
go.mod
2
go.mod
|
@ -7,7 +7,7 @@ require (
|
|||
github.com/Masterminds/sprig/v3 v3.2.2
|
||||
github.com/alecthomas/chroma v0.10.0
|
||||
github.com/aryann/difflib v0.0.0-20210328193216-ff5ff6dc229b
|
||||
github.com/caddyserver/certmagic v0.16.3
|
||||
github.com/caddyserver/certmagic v0.17.0
|
||||
github.com/dustin/go-humanize v1.0.1-0.20200219035652-afde56e7acac
|
||||
github.com/go-chi/chi v4.1.2+incompatible
|
||||
github.com/google/cel-go v0.12.4
|
||||
|
|
4
go.sum
4
go.sum
|
@ -131,8 +131,8 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
|
|||
github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
|
||||
github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
|
||||
github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
|
||||
github.com/caddyserver/certmagic v0.16.3 h1:1ZbiU7y5X0MnDjBTXywUbPMs/ScHbgCeeCy/LPh4IZk=
|
||||
github.com/caddyserver/certmagic v0.16.3/go.mod h1:pSS2aZcdKlrTZrb2DKuRafckx20o5Fz1EdDKEB8KOQM=
|
||||
github.com/caddyserver/certmagic v0.17.0 h1:AHHvvmv6SNcq0vK5BgCevQqYMV8GNprVk6FWZzx8d+Q=
|
||||
github.com/caddyserver/certmagic v0.17.0/go.mod h1:pSS2aZcdKlrTZrb2DKuRafckx20o5Fz1EdDKEB8KOQM=
|
||||
github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=
|
||||
github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
|
||||
github.com/cenkalti/backoff/v3 v3.0.0/go.mod h1:cIeZDE3IrqwwJl6VUwCN6trj1oXrTS4rc0ij+ULvLYs=
|
||||
|
|
367
modules/caddyevents/app.go
Normal file
367
modules/caddyevents/app.go
Normal file
|
@ -0,0 +1,367 @@
|
|||
// Copyright 2015 Matthew Holt and The Caddy 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 caddyevents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/caddyserver/caddy/v2"
|
||||
"github.com/google/uuid"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
func init() {
|
||||
caddy.RegisterModule(App{})
|
||||
}
|
||||
|
||||
// App implements a global eventing system within Caddy.
|
||||
// Modules can emit and subscribe to events, providing
|
||||
// hooks into deep parts of the code base that aren't
|
||||
// otherwise accessible. Events provide information about
|
||||
// what and when things are happening, and this facility
|
||||
// allows handlers to take action when events occur,
|
||||
// add information to the event's metadata, and even
|
||||
// control program flow in some cases.
|
||||
//
|
||||
// Events are propagated in a DOM-like fashion. An event
|
||||
// emitted from module `a.b.c` (the "origin") will first
|
||||
// invoke handlers listening to `a.b.c`, then `a.b`,
|
||||
// then `a`, then those listening regardless of origin.
|
||||
// If a handler returns the special error Aborted, then
|
||||
// propagation immediately stops and the event is marked
|
||||
// as aborted. Emitters may optionally choose to adjust
|
||||
// program flow based on an abort.
|
||||
//
|
||||
// Modules can subscribe to events by origin and/or name.
|
||||
// A handler is invoked only if it is subscribed to the
|
||||
// event by name and origin. Subscriptions should be
|
||||
// registered during the provisioning phase, before apps
|
||||
// are started.
|
||||
//
|
||||
// Event handlers are fired synchronously as part of the
|
||||
// regular flow of the program. This allows event handlers
|
||||
// to control the flow of the program if the origin permits
|
||||
// it and also allows handlers to convey new information
|
||||
// back into the origin module before it continues.
|
||||
// In essence, event handlers are similar to HTTP
|
||||
// middleware handlers.
|
||||
//
|
||||
// Event bindings/subscribers are unordered; i.e.
|
||||
// event handlers are invoked in an arbitrary order.
|
||||
// Event handlers should not rely on the logic of other
|
||||
// handlers to succeed.
|
||||
//
|
||||
// The entirety of this app module is EXPERIMENTAL and
|
||||
// subject to change. Pay attention to release notes.
|
||||
type App struct {
|
||||
// Subscriptions bind handlers to one or more events
|
||||
// either globally or scoped to specific modules or module
|
||||
// namespaces.
|
||||
Subscriptions []*Subscription `json:"subscriptions,omitempty"`
|
||||
|
||||
// Map of event name to map of module ID/namespace to handlers
|
||||
subscriptions map[string]map[caddy.ModuleID][]Handler
|
||||
|
||||
logger *zap.Logger
|
||||
started bool
|
||||
}
|
||||
|
||||
// Subscription represents binding of one or more handlers to
|
||||
// one or more events.
|
||||
type Subscription struct {
|
||||
// The name(s) of the event(s) to bind to. Default: all events.
|
||||
Events []string `json:"events,omitempty"`
|
||||
|
||||
// The ID or namespace of the module(s) from which events
|
||||
// originate to listen to for events. Default: all modules.
|
||||
//
|
||||
// Events propagate up, so events emitted by module "a.b.c"
|
||||
// will also trigger the event for "a.b" and "a". Thus, to
|
||||
// receive all events from "a.b.c" and "a.b.d", for example,
|
||||
// one can subscribe to either "a.b" or all of "a" entirely.
|
||||
Modules []caddy.ModuleID `json:"modules,omitempty"`
|
||||
|
||||
// The event handler modules. These implement the actual
|
||||
// behavior to invoke when an event occurs. At least one
|
||||
// handler is required.
|
||||
HandlersRaw []json.RawMessage `json:"handlers,omitempty" caddy:"namespace=events.handlers inline_key=handler"`
|
||||
|
||||
// The decoded handlers; Go code that is subscribing to
|
||||
// an event should set this field directly; HandlersRaw
|
||||
// is meant for JSON configuration to fill out this field.
|
||||
Handlers []Handler `json:"-"`
|
||||
}
|
||||
|
||||
// CaddyModule returns the Caddy module information.
|
||||
func (App) CaddyModule() caddy.ModuleInfo {
|
||||
return caddy.ModuleInfo{
|
||||
ID: "events",
|
||||
New: func() caddy.Module { return new(App) },
|
||||
}
|
||||
}
|
||||
|
||||
// Provision sets up the app.
|
||||
func (app *App) Provision(ctx caddy.Context) error {
|
||||
app.logger = ctx.Logger(app)
|
||||
app.subscriptions = make(map[string]map[caddy.ModuleID][]Handler)
|
||||
|
||||
for _, sub := range app.Subscriptions {
|
||||
if sub.HandlersRaw != nil {
|
||||
handlersIface, err := ctx.LoadModule(sub, "HandlersRaw")
|
||||
if err != nil {
|
||||
return fmt.Errorf("loading event subscriber modules: %v", err)
|
||||
}
|
||||
for _, h := range handlersIface.([]any) {
|
||||
sub.Handlers = append(sub.Handlers, h.(Handler))
|
||||
}
|
||||
if len(sub.Handlers) == 0 {
|
||||
// pointless to bind without any handlers
|
||||
return fmt.Errorf("no handlers defined")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Start runs the app.
|
||||
func (app *App) Start() error {
|
||||
for _, sub := range app.Subscriptions {
|
||||
if err := app.Subscribe(sub); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
app.started = true
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop gracefully shuts down the app.
|
||||
func (app *App) Stop() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Subscribe binds one or more event handlers to one or more events
|
||||
// according to the subscription s. For now, subscriptions can only
|
||||
// be created during the provision phase; new bindings cannot be
|
||||
// created after the events app has started.
|
||||
func (app *App) Subscribe(s *Subscription) error {
|
||||
if app.started {
|
||||
return fmt.Errorf("events already started; new subscriptions closed")
|
||||
}
|
||||
|
||||
// handle special case of catch-alls (omission of event name or module space implies all)
|
||||
if len(s.Events) == 0 {
|
||||
s.Events = []string{""}
|
||||
}
|
||||
if len(s.Modules) == 0 {
|
||||
s.Modules = []caddy.ModuleID{""}
|
||||
}
|
||||
|
||||
for _, eventName := range s.Events {
|
||||
if app.subscriptions[eventName] == nil {
|
||||
app.subscriptions[eventName] = make(map[caddy.ModuleID][]Handler)
|
||||
}
|
||||
for _, originModule := range s.Modules {
|
||||
app.subscriptions[eventName][originModule] = append(app.subscriptions[eventName][originModule], s.Handlers...)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// On is syntactic sugar for Subscribe() that binds a single handler
|
||||
// to a single event from any module. If the eventName is empty string,
|
||||
// it counts for all events.
|
||||
func (app *App) On(eventName string, handler Handler) error {
|
||||
return app.Subscribe(&Subscription{
|
||||
Events: []string{eventName},
|
||||
Handlers: []Handler{handler},
|
||||
})
|
||||
}
|
||||
|
||||
// Emit creates and dispatches an event named eventName to all relevant handlers with
|
||||
// the metadata data. Events are emitted and propagated synchronously. The returned Event
|
||||
// value will have any additional information from the invoked handlers.
|
||||
func (app *App) Emit(ctx caddy.Context, eventName string, data map[string]any) Event {
|
||||
id, err := uuid.NewRandom()
|
||||
if err != nil {
|
||||
app.logger.Error("failed generating new event ID",
|
||||
zap.Error(err),
|
||||
zap.String("event", eventName))
|
||||
}
|
||||
|
||||
eventName = strings.ToLower(eventName)
|
||||
|
||||
e := Event{
|
||||
id: id,
|
||||
ts: time.Now(),
|
||||
name: eventName,
|
||||
origin: ctx.Module(),
|
||||
data: data,
|
||||
}
|
||||
|
||||
// add event info to replacer, make sure it's in the context
|
||||
repl, ok := ctx.Context.Value(caddy.ReplacerCtxKey).(*caddy.Replacer)
|
||||
if !ok {
|
||||
repl = caddy.NewReplacer()
|
||||
ctx.Context = context.WithValue(ctx.Context, caddy.ReplacerCtxKey, repl)
|
||||
}
|
||||
repl.Map(func(key string) (any, bool) {
|
||||
switch key {
|
||||
case "event":
|
||||
return e, true
|
||||
case "event.id":
|
||||
return e.id, true
|
||||
case "event.name":
|
||||
return e.name, true
|
||||
case "event.time":
|
||||
return e.ts, true
|
||||
case "event.time_unix":
|
||||
return e.ts.UnixMilli(), true
|
||||
case "event.module":
|
||||
return e.origin.CaddyModule().ID, true
|
||||
case "event.data":
|
||||
return e.data, true
|
||||
}
|
||||
|
||||
if strings.HasPrefix(key, "event.data.") {
|
||||
key = strings.TrimPrefix(key, "event.data.")
|
||||
if val, ok := data[key]; ok {
|
||||
return val, true
|
||||
}
|
||||
}
|
||||
|
||||
return nil, false
|
||||
})
|
||||
|
||||
app.logger.Debug("event",
|
||||
zap.String("name", e.name),
|
||||
zap.String("id", e.id.String()),
|
||||
zap.String("origin", e.origin.CaddyModule().String()),
|
||||
zap.Any("data", e.data),
|
||||
)
|
||||
|
||||
// invoke handlers bound to the event by name and also all events; this for loop
|
||||
// iterates twice at most: once for the event name, once for "" (all events)
|
||||
for {
|
||||
moduleID := e.origin.CaddyModule().ID
|
||||
|
||||
// implement propagation up the module tree (i.e. start with "a.b.c" then "a.b" then "a" then "")
|
||||
for {
|
||||
if app.subscriptions[eventName] == nil {
|
||||
break // shortcut if event not bound at all
|
||||
}
|
||||
|
||||
for _, handler := range app.subscriptions[eventName][moduleID] {
|
||||
if err := handler.Handle(ctx, e); err != nil {
|
||||
aborted := errors.Is(err, ErrAborted)
|
||||
|
||||
app.logger.Error("handler error",
|
||||
zap.Error(err),
|
||||
zap.Bool("aborted", aborted))
|
||||
|
||||
if aborted {
|
||||
e.Aborted = err
|
||||
return e
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if moduleID == "" {
|
||||
break
|
||||
}
|
||||
lastDot := strings.LastIndex(string(moduleID), ".")
|
||||
if lastDot < 0 {
|
||||
moduleID = "" // include handlers bound to events regardless of module
|
||||
} else {
|
||||
moduleID = moduleID[:lastDot]
|
||||
}
|
||||
}
|
||||
|
||||
// include handlers listening to all events
|
||||
if eventName == "" {
|
||||
break
|
||||
}
|
||||
eventName = ""
|
||||
}
|
||||
|
||||
return e
|
||||
}
|
||||
|
||||
// Event represents something that has happened or is happening.
|
||||
type Event struct {
|
||||
id uuid.UUID
|
||||
ts time.Time
|
||||
name string
|
||||
origin caddy.Module
|
||||
data map[string]any
|
||||
|
||||
// If non-nil, the event has been aborted, meaning
|
||||
// propagation has stopped to other handlers and
|
||||
// the code should stop what it was doing. Emitters
|
||||
// may choose to use this as a signal to adjust their
|
||||
// code path appropriately.
|
||||
Aborted error
|
||||
}
|
||||
|
||||
// CloudEvent exports event e as a structure that, when
|
||||
// serialized as JSON, is compatible with the
|
||||
// CloudEvents spec.
|
||||
func (e Event) CloudEvent() CloudEvent {
|
||||
dataJSON, _ := json.Marshal(e.data)
|
||||
return CloudEvent{
|
||||
ID: e.id.String(),
|
||||
Source: e.origin.CaddyModule().String(),
|
||||
SpecVersion: "1.0",
|
||||
Type: e.name,
|
||||
Time: e.ts,
|
||||
DataContentType: "application/json",
|
||||
Data: dataJSON,
|
||||
}
|
||||
}
|
||||
|
||||
// CloudEvent is a JSON-serializable structure that
|
||||
// is compatible with the CloudEvents specification.
|
||||
// See https://cloudevents.io.
|
||||
type CloudEvent struct {
|
||||
ID string `json:"id"`
|
||||
Source string `json:"source"`
|
||||
SpecVersion string `json:"specversion"`
|
||||
Type string `json:"type"`
|
||||
Time time.Time `json:"time"`
|
||||
DataContentType string `json:"datacontenttype,omitempty"`
|
||||
Data json.RawMessage `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
// ErrAborted cancels an event.
|
||||
var ErrAborted = errors.New("event aborted")
|
||||
|
||||
// Handler is a type that can handle events.
|
||||
type Handler interface {
|
||||
Handle(context.Context, Event) error
|
||||
}
|
||||
|
||||
// Interface guards
|
||||
var (
|
||||
_ caddy.App = (*App)(nil)
|
||||
_ caddy.Provisioner = (*App)(nil)
|
||||
)
|
88
modules/caddyevents/eventsconfig/caddyfile.go
Normal file
88
modules/caddyevents/eventsconfig/caddyfile.go
Normal file
|
@ -0,0 +1,88 @@
|
|||
// Copyright 2015 Matthew Holt and The Caddy 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 eventsconfig is for configuring caddyevents.App with the
|
||||
// Caddyfile. This code can't be in the caddyevents package because
|
||||
// the httpcaddyfile package imports caddyhttp, which imports
|
||||
// caddyevents: hence, it creates an import cycle.
|
||||
package eventsconfig
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"github.com/caddyserver/caddy/v2/caddyconfig"
|
||||
"github.com/caddyserver/caddy/v2/caddyconfig/caddyfile"
|
||||
"github.com/caddyserver/caddy/v2/caddyconfig/httpcaddyfile"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyevents"
|
||||
)
|
||||
|
||||
func init() {
|
||||
httpcaddyfile.RegisterGlobalOption("events", parseApp)
|
||||
}
|
||||
|
||||
// parseApp configures the "events" global option from Caddyfile to set up the events app.
|
||||
// Syntax:
|
||||
//
|
||||
// events {
|
||||
// on <event> <handler_module...>
|
||||
// }
|
||||
//
|
||||
// If <event> is *, then it will bind to all events.
|
||||
func parseApp(d *caddyfile.Dispenser, _ any) (any, error) {
|
||||
app := new(caddyevents.App)
|
||||
|
||||
// consume the option name
|
||||
if !d.Next() {
|
||||
return nil, d.ArgErr()
|
||||
}
|
||||
|
||||
// handle the block
|
||||
for d.NextBlock(0) {
|
||||
switch d.Val() {
|
||||
case "on":
|
||||
if !d.NextArg() {
|
||||
return nil, d.ArgErr()
|
||||
}
|
||||
eventName := d.Val()
|
||||
if eventName == "*" {
|
||||
eventName = ""
|
||||
}
|
||||
|
||||
if !d.NextArg() {
|
||||
return nil, d.ArgErr()
|
||||
}
|
||||
handlerName := d.Val()
|
||||
modID := "events.handlers." + handlerName
|
||||
unm, err := caddyfile.UnmarshalModule(d, modID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
app.Subscriptions = append(app.Subscriptions, &caddyevents.Subscription{
|
||||
Events: []string{eventName},
|
||||
HandlersRaw: []json.RawMessage{
|
||||
caddyconfig.JSONModuleObject(unm, "handler", handlerName, nil),
|
||||
},
|
||||
})
|
||||
|
||||
default:
|
||||
return nil, d.ArgErr()
|
||||
}
|
||||
}
|
||||
|
||||
return httpcaddyfile.App{
|
||||
Name: "events",
|
||||
Value: caddyconfig.JSON(app, nil),
|
||||
}, nil
|
||||
}
|
|
@ -24,6 +24,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/caddyserver/caddy/v2"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyevents"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddytls"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/net/http2"
|
||||
|
@ -161,6 +162,11 @@ func (app *App) Provision(ctx caddy.Context) error {
|
|||
app.ctx = ctx
|
||||
app.logger = ctx.Logger(app)
|
||||
|
||||
eventsAppIface, err := ctx.App("events")
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting events app: %v", err)
|
||||
}
|
||||
|
||||
repl := caddy.NewReplacer()
|
||||
|
||||
// this provisions the matchers for each route,
|
||||
|
@ -175,6 +181,8 @@ func (app *App) Provision(ctx caddy.Context) error {
|
|||
for srvName, srv := range app.Servers {
|
||||
srv.name = srvName
|
||||
srv.tlsApp = app.tlsApp
|
||||
srv.events = eventsAppIface.(*caddyevents.App)
|
||||
srv.ctx = ctx
|
||||
srv.logger = app.logger.Named("log")
|
||||
srv.errorLogger = app.logger.Named("log.error")
|
||||
srv.shutdownAtMu = new(sync.RWMutex)
|
||||
|
|
|
@ -284,6 +284,13 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, upstre
|
|||
}
|
||||
}
|
||||
|
||||
markUnhealthy := func() {
|
||||
// dispatch an event that the host newly became unhealthy
|
||||
if upstream.setHealthy(false) {
|
||||
h.events.Emit(h.ctx, "unhealthy", map[string]any{"host": hostAddr})
|
||||
}
|
||||
}
|
||||
|
||||
// do the request, being careful to tame the response body
|
||||
resp, err := h.HealthChecks.Active.httpClient.Do(req)
|
||||
if err != nil {
|
||||
|
@ -291,7 +298,7 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, upstre
|
|||
zap.String("host", hostAddr),
|
||||
zap.Error(err),
|
||||
)
|
||||
upstream.setHealthy(false)
|
||||
markUnhealthy()
|
||||
return nil
|
||||
}
|
||||
var body io.Reader = resp.Body
|
||||
|
@ -311,7 +318,7 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, upstre
|
|||
zap.Int("status_code", resp.StatusCode),
|
||||
zap.String("host", hostAddr),
|
||||
)
|
||||
upstream.setHealthy(false)
|
||||
markUnhealthy()
|
||||
return nil
|
||||
}
|
||||
} else if resp.StatusCode < 200 || resp.StatusCode >= 400 {
|
||||
|
@ -319,7 +326,7 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, upstre
|
|||
zap.Int("status_code", resp.StatusCode),
|
||||
zap.String("host", hostAddr),
|
||||
)
|
||||
upstream.setHealthy(false)
|
||||
markUnhealthy()
|
||||
return nil
|
||||
}
|
||||
|
||||
|
@ -331,14 +338,14 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, upstre
|
|||
zap.String("host", hostAddr),
|
||||
zap.Error(err),
|
||||
)
|
||||
upstream.setHealthy(false)
|
||||
markUnhealthy()
|
||||
return nil
|
||||
}
|
||||
if !h.HealthChecks.Active.bodyRegexp.Match(bodyBytes) {
|
||||
h.HealthChecks.Active.logger.Info("response body failed expectations",
|
||||
zap.String("host", hostAddr),
|
||||
)
|
||||
upstream.setHealthy(false)
|
||||
markUnhealthy()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
@ -346,6 +353,7 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, upstre
|
|||
// passed health check parameters, so mark as healthy
|
||||
if upstream.setHealthy(true) {
|
||||
h.HealthChecks.Active.logger.Info("host is up", zap.String("host", hostAddr))
|
||||
h.events.Emit(h.ctx, "healthy", map[string]any{"host": hostAddr})
|
||||
}
|
||||
|
||||
return nil
|
||||
|
|
|
@ -36,6 +36,7 @@ import (
|
|||
|
||||
"github.com/caddyserver/caddy/v2"
|
||||
"github.com/caddyserver/caddy/v2/caddyconfig/caddyfile"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyevents"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyhttp"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyhttp/headers"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyhttp/rewrite"
|
||||
|
@ -193,6 +194,7 @@ type Handler struct {
|
|||
|
||||
ctx caddy.Context
|
||||
logger *zap.Logger
|
||||
events *caddyevents.App
|
||||
}
|
||||
|
||||
// CaddyModule returns the Caddy module information.
|
||||
|
@ -205,6 +207,11 @@ func (Handler) CaddyModule() caddy.ModuleInfo {
|
|||
|
||||
// Provision ensures that h is set up properly before use.
|
||||
func (h *Handler) Provision(ctx caddy.Context) error {
|
||||
eventAppIface, err := ctx.App("events")
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting events app: %v", err)
|
||||
}
|
||||
h.events = eventAppIface.(*caddyevents.App)
|
||||
h.ctx = ctx
|
||||
h.logger = ctx.Logger(h)
|
||||
|
||||
|
|
|
@ -30,6 +30,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/caddyserver/caddy/v2"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyevents"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddytls"
|
||||
"github.com/caddyserver/certmagic"
|
||||
"github.com/lucas-clemente/quic-go/http3"
|
||||
|
@ -154,9 +155,11 @@ type Server struct {
|
|||
listeners []net.Listener
|
||||
|
||||
tlsApp *caddytls.TLS
|
||||
events *caddyevents.App
|
||||
logger *zap.Logger
|
||||
accessLogger *zap.Logger
|
||||
errorLogger *zap.Logger
|
||||
ctx caddy.Context
|
||||
|
||||
server *http.Server
|
||||
h3server *http3.Server
|
||||
|
|
|
@ -256,6 +256,7 @@ func (ap *AutomationPolicy) Provision(tlsApp *TLS) error {
|
|||
MustStaple: ap.MustStaple,
|
||||
RenewalWindowRatio: ap.RenewalWindowRatio,
|
||||
KeySource: keySource,
|
||||
OnEvent: tlsApp.onEvent,
|
||||
OnDemand: ond,
|
||||
OCSP: certmagic.OCSPConfig{
|
||||
DisableStapling: ap.DisableOCSPStapling,
|
||||
|
|
|
@ -15,6 +15,7 @@
|
|||
package caddytls
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
@ -25,6 +26,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/caddyserver/caddy/v2"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyevents"
|
||||
"github.com/caddyserver/certmagic"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
@ -73,6 +75,7 @@ type TLS struct {
|
|||
storageCleanTicker *time.Ticker
|
||||
storageCleanStop chan struct{}
|
||||
logger *zap.Logger
|
||||
events *caddyevents.App
|
||||
}
|
||||
|
||||
// CaddyModule returns the Caddy module information.
|
||||
|
@ -85,6 +88,11 @@ func (TLS) CaddyModule() caddy.ModuleInfo {
|
|||
|
||||
// Provision sets up the configuration for the TLS app.
|
||||
func (t *TLS) Provision(ctx caddy.Context) error {
|
||||
eventsAppIface, err := ctx.App("events")
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting events app: %v", err)
|
||||
}
|
||||
t.events = eventsAppIface.(*caddyevents.App)
|
||||
t.ctx = ctx
|
||||
t.logger = ctx.Logger(t)
|
||||
repl := caddy.NewReplacer()
|
||||
|
@ -189,6 +197,7 @@ func (t *TLS) Provision(ctx caddy.Context) error {
|
|||
magic := certmagic.New(t.certCache, certmagic.Config{
|
||||
Storage: ctx.Storage(),
|
||||
Logger: t.logger,
|
||||
OnEvent: t.onEvent,
|
||||
OCSP: certmagic.OCSPConfig{
|
||||
DisableStapling: t.DisableOCSPStapling,
|
||||
},
|
||||
|
@ -514,6 +523,12 @@ func (t *TLS) storageCleanInterval() time.Duration {
|
|||
return defaultStorageCleanInterval
|
||||
}
|
||||
|
||||
// onEvent translates CertMagic events into Caddy events then dispatches them.
|
||||
func (t *TLS) onEvent(ctx context.Context, eventName string, data map[string]any) error {
|
||||
evt := t.events.Emit(t.ctx, eventName, data)
|
||||
return evt.Aborted
|
||||
}
|
||||
|
||||
// CertificateLoader is a type that can load certificates.
|
||||
// Certificates can optionally be associated with tags.
|
||||
type CertificateLoader interface {
|
||||
|
|
|
@ -3,6 +3,8 @@ package standard
|
|||
import (
|
||||
// standard Caddy modules
|
||||
_ "github.com/caddyserver/caddy/v2/caddyconfig/caddyfile"
|
||||
_ "github.com/caddyserver/caddy/v2/modules/caddyevents"
|
||||
_ "github.com/caddyserver/caddy/v2/modules/caddyevents/eventsconfig"
|
||||
_ "github.com/caddyserver/caddy/v2/modules/caddyhttp/standard"
|
||||
_ "github.com/caddyserver/caddy/v2/modules/caddypki"
|
||||
_ "github.com/caddyserver/caddy/v2/modules/caddypki/acmeserver"
|
||||
|
|
20
usagepool.go
20
usagepool.go
|
@ -25,15 +25,15 @@ import (
|
|||
// only inserted if they do not already exist. There
|
||||
// are two ways to add values to the pool:
|
||||
//
|
||||
// 1) LoadOrStore will increment usage and store the
|
||||
// value immediately if it does not already exist.
|
||||
// 2) LoadOrNew will atomically check for existence
|
||||
// and construct the value immediately if it does
|
||||
// not already exist, or increment the usage
|
||||
// otherwise, then store that value in the pool.
|
||||
// When the constructed value is finally deleted
|
||||
// from the pool (when its usage reaches 0), it
|
||||
// will be cleaned up by calling Destruct().
|
||||
// 1. LoadOrStore will increment usage and store the
|
||||
// value immediately if it does not already exist.
|
||||
// 2. LoadOrNew will atomically check for existence
|
||||
// and construct the value immediately if it does
|
||||
// not already exist, or increment the usage
|
||||
// otherwise, then store that value in the pool.
|
||||
// When the constructed value is finally deleted
|
||||
// from the pool (when its usage reaches 0), it
|
||||
// will be cleaned up by calling Destruct().
|
||||
//
|
||||
// The use of LoadOrNew allows values to be created
|
||||
// and reused and finally cleaned up only once, even
|
||||
|
@ -196,7 +196,7 @@ func (up *UsagePool) Delete(key any) (deleted bool, err error) {
|
|||
|
||||
// References returns the number of references (count of usages) to a
|
||||
// key in the pool, and true if the key exists, or false otherwise.
|
||||
func (up *UsagePool) References(key interface{}) (int, bool) {
|
||||
func (up *UsagePool) References(key any) (int, bool) {
|
||||
up.RLock()
|
||||
upv, loaded := up.pool[key]
|
||||
up.RUnlock()
|
||||
|
|
Loading…
Reference in New Issue
Block a user