242 lines
7.1 KiB
Go
242 lines
7.1 KiB
Go
package web
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/aws/arn"
|
|
"github.com/gin-contrib/sessions"
|
|
"github.com/gin-gonic/gin"
|
|
|
|
"code.osinet.fr/fgm/sqs_demo/back/services/redriver"
|
|
)
|
|
|
|
// classifyPolicies distributes queue policies across five bins:
|
|
// - wild: policies describing a DLQ with an allowAll policy
|
|
// - closed: policies describing a DLQ with an denyAll policy
|
|
// - byQueue: policies describing a DLQ with an byQueue policy
|
|
// - sources: policies describing a queue with a DLQ
|
|
// - notSources: policies describing a queue without a DLQ
|
|
//
|
|
// Note that dead-letter queues may also be source queues, for multi-level processing, and vice-versa.
|
|
//
|
|
// For easier reading, DLQs are not listed as part of the non-source list.
|
|
func classifyPolicies(policies map[string]*redriver.QueueRedrivePolicies) (byQueue, wild, closed, sources map[string]redriver.QueueRedrivePolicies, notSources []string) {
|
|
DLQs := make(map[string]redriver.QueueRedrivePolicies)
|
|
sources = make(map[string]redriver.QueueRedrivePolicies)
|
|
wild = make(map[string]redriver.QueueRedrivePolicies)
|
|
closed = make(map[string]redriver.QueueRedrivePolicies)
|
|
byQueue = make(map[string]redriver.QueueRedrivePolicies)
|
|
nsTemp := make([]string, 0, len(notSources))
|
|
|
|
for qURL, qrp := range policies {
|
|
if qrp.QueueInfoAttributesRedriveAllowPolicy != nil {
|
|
DLQs[qURL] = *qrp
|
|
}
|
|
if qrp.QueueInfoAttributesRedrivePolicy != nil {
|
|
sources[qURL] = *qrp
|
|
} else {
|
|
nsTemp = append(nsTemp, qURL)
|
|
}
|
|
}
|
|
for _, qURL := range nsTemp {
|
|
if _, ok := DLQs[qURL]; !ok {
|
|
notSources = append(notSources, qURL)
|
|
}
|
|
}
|
|
for qURL, qrp := range DLQs {
|
|
rap := qrp.QueueInfoAttributesRedriveAllowPolicy
|
|
switch rap.RedrivePermission {
|
|
case "allowAll":
|
|
wild[qURL] = qrp
|
|
case "denyAll":
|
|
closed[qURL] = qrp
|
|
case "byQueue":
|
|
byQueue[qURL] = qrp
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func policiesByURL(c *gin.Context, qURLs []string, rd redriver.Redriver, ctx context.Context) map[string]*redriver.QueueRedrivePolicies {
|
|
qMap := make(map[string]*redriver.QueueRedrivePolicies, len(qURLs))
|
|
for _, qURL := range qURLs {
|
|
name, err := redriver.NameFromURL(qURL)
|
|
if err != nil {
|
|
log.Println(http.StatusInternalServerError)
|
|
c.HTML(http.StatusInternalServerError, "500", nil)
|
|
}
|
|
qrp, err := rd.GetRedrivePolicies(ctx, name)
|
|
if err != nil || qrp == nil {
|
|
log.Println(err)
|
|
c.HTML(http.StatusInternalServerError, "500", nil)
|
|
}
|
|
qMap[qURL] = qrp
|
|
}
|
|
return qMap
|
|
}
|
|
|
|
type link struct{ URL, Text string }
|
|
|
|
// queueRow provides templates with a representation for all kinds of table cells
|
|
type queueCell struct {
|
|
Message string
|
|
Link *link
|
|
Links [][]link
|
|
}
|
|
|
|
type QueueRow [2]queueCell
|
|
|
|
func makeHomeHandler(rd redriver.Redriver, prefix string) gin.HandlerFunc {
|
|
return func(c *gin.Context) {
|
|
ctx := c.Request.Context()
|
|
sess := sessions.Default(c)
|
|
flashes := sess.Flashes()
|
|
defer func() { _ = sess.Save() }()
|
|
|
|
t0 := time.Now()
|
|
qURLs, err := rd.ListQueues(ctx, prefix)
|
|
latency := time.Since(t0)
|
|
if err != nil {
|
|
log.Printf("failed listing queues: %v", err)
|
|
c.HTML(http.StatusInternalServerError, "500", nil)
|
|
return
|
|
}
|
|
policies := policiesByURL(c, qURLs, rd, ctx)
|
|
byQueue, wild, closed, sources, notSources := classifyPolicies(policies)
|
|
// log.Printf("DLQs: %#v\nSRCs: %#v\nNoSRCs: %#v\n", maps.Keys(DLQs), maps.Keys(SRCs), NoSRCs)
|
|
rows := make([]QueueRow, 0, len(byQueue)+3) // wild, closed, notSources: sources appear within a byQueue row
|
|
|
|
rows, err = prepareSingleLeft(rows, wild, "All source queues allowed")
|
|
if err != nil {
|
|
log.Printf("failed preparing wild DLQs: %v", err)
|
|
c.HTML(http.StatusInternalServerError, "500", nil)
|
|
return
|
|
}
|
|
rows, err = prepareSingleLeft(rows, closed, "No source queue allowed")
|
|
if err != nil {
|
|
log.Printf("failed preparing closed DLQs: %v", err)
|
|
c.HTML(http.StatusInternalServerError, "500", nil)
|
|
return
|
|
}
|
|
rows, err = prepareSingleRight(rows, notSources, "No associated DLQ")
|
|
if err != nil {
|
|
log.Printf("failed converting isolated queues: %v", err)
|
|
c.HTML(http.StatusInternalServerError, "500", nil)
|
|
return
|
|
}
|
|
rows, err = prepareBindings(rows, byQueue, sources)
|
|
if err != nil {
|
|
log.Printf("failed preparing queue bindings: %v", err)
|
|
c.HTML(http.StatusInternalServerError, "500", nil)
|
|
return
|
|
}
|
|
c.HTML(http.StatusOK, "home", gin.H{
|
|
"flashes": flashes,
|
|
"latency": latency,
|
|
"prefix": prefix,
|
|
"rows": rows,
|
|
})
|
|
|
|
}
|
|
}
|
|
|
|
func prepareSingleLeft(rows []QueueRow, list map[string]redriver.QueueRedrivePolicies, msg string) ([]QueueRow, error) {
|
|
if len(list) > 0 {
|
|
row := QueueRow{
|
|
{Message: msg},
|
|
{Links: make([][]link, 1)},
|
|
}
|
|
row[1].Links[0] = make([]link, 0, len(list))
|
|
for qURL := range list {
|
|
name, err := redriver.NameFromURL(qURL)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
row[1].Links[0] = append(row[1].Links[0], link{URL: "/queue/" + name, Text: name})
|
|
}
|
|
rows = append(rows, row)
|
|
}
|
|
return rows, nil
|
|
}
|
|
|
|
func prepareSingleRight(rows []QueueRow, list []string, msg string) ([]QueueRow, error) {
|
|
if len(list) > 0 {
|
|
row := QueueRow{
|
|
{Links: make([][]link, 1)},
|
|
{Message: msg},
|
|
}
|
|
row[0].Links[0] = make([]link, 0, len(list))
|
|
for _, qURL := range list {
|
|
name, err := redriver.NameFromURL(qURL)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
row[0].Links[0] = append(row[0].Links[0], link{URL: "/queue/" + name, Text: name})
|
|
}
|
|
rows = append(rows, row)
|
|
}
|
|
return rows, nil
|
|
}
|
|
|
|
func prepareBinding(dlqURL string, DLQrp redriver.QueueRedrivePolicies, sources map[string]redriver.QueueRedrivePolicies) (*QueueRow, error) {
|
|
var left, right queueCell
|
|
dlqName, err := redriver.NameFromURL(dlqURL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed parsing DLQ URL %q: %w", dlqURL, err)
|
|
}
|
|
right = queueCell{Link: &link{URL: "/queue/" + dlqName, Text: dlqName}}
|
|
|
|
a, err := redriver.ARNFromURL(dlqURL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed converting URL %q to ARN: %w", dlqURL, err)
|
|
}
|
|
dlqARN := a.String()
|
|
var bound, unbound, elsewhere []link
|
|
for _, sourceARN := range DLQrp.SourceQueueARNs {
|
|
sa, err := arn.Parse(sourceARN)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed parsing %q as an ARN: %w", sourceARN, err)
|
|
}
|
|
sourceURL, err := redriver.URLFromARN(sa)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed converting ARN %q for URL: %w", sa.String(), err)
|
|
}
|
|
sourceName := sa.Resource
|
|
sourceQRP, ok := sources[sourceURL]
|
|
link := link{
|
|
URL: "/queue/" + sourceName,
|
|
Text: sourceName,
|
|
}
|
|
if ok {
|
|
switch sourceQRP.DeadLetterTargetARN {
|
|
case "":
|
|
unbound = append(unbound, link)
|
|
case dlqARN:
|
|
bound = append(bound, link)
|
|
default:
|
|
elsewhere = append(elsewhere, link)
|
|
}
|
|
} else {
|
|
unbound = append(unbound, link)
|
|
}
|
|
}
|
|
left.Links = [][]link{bound, elsewhere, unbound}
|
|
|
|
return &QueueRow{left, right}, nil
|
|
}
|
|
|
|
func prepareBindings(rows []QueueRow, DLQs, sources map[string]redriver.QueueRedrivePolicies) ([]QueueRow, error) {
|
|
for dlqURL, DLQrp := range DLQs {
|
|
row, err := prepareBinding(dlqURL, DLQrp, sources)
|
|
if err != nil || row == nil {
|
|
return nil, fmt.Errorf("failed binding DLQ %s: %w", dlqURL, err)
|
|
}
|
|
rows = append(rows, *row)
|
|
}
|
|
return rows, nil
|
|
}
|